如何在Python中获得类似调度程序的Cron ?

我正在寻找Python中的一个库，它将提供at和cron之类的功能。

我非常希望有一个纯Python的解决方案，而不是依赖于安装在盒子上的工具;这样我在没有cron的机器上运行。

对于那些不熟悉cron的人:你可以基于一个表达式来调度任务，比如:

 0 2 * * 7 /usr/bin/run-backup # run the backups at 0200 on Every Sunday
0 9-17/2 * * 1-5 /usr/bin/purge-temps # run the purge temps command, every 2 hours between 9am and 5pm on Mondays to Fridays.

cron时间表达式语法不那么重要，但我希望具有这种灵活性。

如果没有什么东西可以为我开箱即用，任何关于构建模块的建议都将被感激地接受。

<强>编辑 我对启动进程不感兴趣，只对同样用Python编写的“作业”——Python函数感兴趣。根据需要，我认为这将是一个不同的线程，但不是在不同的进程

为此，我正在寻找cron time表达式的表达性，但在Python中。

Cron 有已经存在多年了，但我正试图尽可能地可移植。我不能指望它的存在。

405881 次浏览
小开

我不知道这样的东西是否已经存在。编写自己的time, datetime和/或calendar模块很容易，参见http://docs.python.org/library/time.html

python解决方案的唯一问题是，你的作业需要一直运行，并且可能在重新启动后自动“复活”，为此你做需要依赖于系统相关的解决方案。

小开

没有“纯python”的方法来做到这一点，因为其他一些进程必须启动python才能运行您的解决方案。每个平台都有一种或20种不同的方式来启动流程并监视其进展。在unix平台上，cron是旧的标准。在Mac OS X上也有launchd，它结合了类似cron的启动和看门狗功能，如果你想要的话，可以让你的进程保持活跃。一旦python开始运行，你就可以使用固定播送时间模块来安排任务。

小开

您可以使用普通的Python参数传递语法来指定crontab。例如，假设我们定义一个Event类如下所示:

from datetime import datetime, timedelta
import time


# Some utility classes / functions first
class AllMatch(set):
"""Universal set - match everything"""
def __contains__(self, item): return True


allMatch = AllMatch()


def conv_to_set(obj):  # Allow single integer to be provided
if isinstance(obj, (int,long)):
return set([obj])  # Single item
if not isinstance(obj, set):
obj = set(obj)
return obj


# The actual Event class
class Event(object):
def __init__(self, action, min=allMatch, hour=allMatch,
day=allMatch, month=allMatch, dow=allMatch,
args=(), kwargs={}):
self.mins = conv_to_set(min)
self.hours= conv_to_set(hour)
self.days = conv_to_set(day)
self.months = conv_to_set(month)
self.dow = conv_to_set(dow)
self.action = action
self.args = args
self.kwargs = kwargs


def matchtime(self, t):
"""Return True if this event should trigger at the specified datetime"""
return ((t.minute     in self.mins) and
(t.hour       in self.hours) and
(t.day        in self.days) and
(t.month      in self.months) and
(t.weekday()  in self.dow))


def check(self, t):
if self.matchtime(t):
self.action(*self.args, **self.kwargs)

(注:未经彻底测试)

然后你的CronTab可以在正常的python语法中指定:

c = CronTab(
Event(perform_backup, 0, 2, dow=6 ),
Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)

通过这种方式，您可以获得Python参数机制的全部功能(混合位置和关键字参数，并可以使用符号名称作为周和月的名称)

CronTab类将被定义为简单地按分钟增量休眠，并在每个事件上调用check()。(不过，夏令时/时区可能有一些微妙之处需要警惕)。下面是一个快速实现:

class CronTab(object):
def __init__(self, *events):
self.events = events


def run(self):
t=datetime(*datetime.now().timetuple()[:5])
while 1:
for e in self.events:
e.check(t)


t += timedelta(minutes=1)
while datetime.now() < t:
time.sleep((t - datetime.now()).seconds)

需要注意的几件事:Python的工作日/月份是零索引(不像cron)，并且该范围不包括最后一个元素，因此像“1-5”这样的语法变成了范围(0,5)- ie[0,1,2,3,4]。如果您更喜欢cron语法，解析它应该不会太难。

小开

和上面差不多，但是同时使用gevent:)

"""Gevent based crontab implementation"""


from datetime import datetime, timedelta
import gevent


# Some utility classes / functions first
def conv_to_set(obj):
"""Converts to set allowing single integer to be provided"""


if isinstance(obj, (int, long)):
return set([obj])  # Single item
if not isinstance(obj, set):
obj = set(obj)
return obj


class AllMatch(set):
"""Universal set - match everything"""
def __contains__(self, item):
return True


allMatch = AllMatch()


class Event(object):
"""The Actual Event Class"""


def __init__(self, action, minute=allMatch, hour=allMatch,
day=allMatch, month=allMatch, daysofweek=allMatch,
args=(), kwargs={}):
self.mins = conv_to_set(minute)
self.hours = conv_to_set(hour)
self.days = conv_to_set(day)
self.months = conv_to_set(month)
self.daysofweek = conv_to_set(daysofweek)
self.action = action
self.args = args
self.kwargs = kwargs


def matchtime(self, t1):
"""Return True if this event should trigger at the specified datetime"""
return ((t1.minute     in self.mins) and
(t1.hour       in self.hours) and
(t1.day        in self.days) and
(t1.month      in self.months) and
(t1.weekday()  in self.daysofweek))


def check(self, t):
"""Check and run action if needed"""


if self.matchtime(t):
self.action(*self.args, **self.kwargs)


class CronTab(object):
"""The crontab implementation"""


def __init__(self, *events):
self.events = events


def _check(self):
"""Check all events in separate greenlets"""


t1 = datetime(*datetime.now().timetuple()[:5])
for event in self.events:
gevent.spawn(event.check, t1)


t1 += timedelta(minutes=1)
s1 = (t1 - datetime.now()).seconds + 1
print "Checking again in %s seconds" % s1
job = gevent.spawn_later(s1, self._check)


def run(self):
"""Run the cron forever"""


self._check()
while True:
gevent.sleep(60)


import os
def test_task():
"""Just an example that sends a bell and asd to all terminals"""


os.system('echo asd | wall')


cron = CronTab(
Event(test_task, 22, 1 ),
Event(test_task, 0, range(9,18,2), daysofweek=range(0,5)),
)
cron.run()
小开

另一个简单的解是:

from aqcron import At
from time import sleep
from datetime import datetime


# Event scheduling
event_1 = At( second=5 )
event_2 = At( second=[0,20,40] )


while True:
now = datetime.now()


# Event check
if now in event_1: print "event_1"
if now in event_2: print "event_2"


sleep(1)

和类aqcron。的是:

# aqcron.py


class At(object):
def __init__(self, year=None,    month=None,
day=None,     weekday=None,
hour=None,    minute=None,
second=None):
loc = locals()
loc.pop("self")
self.at = dict((k, v) for k, v in loc.iteritems() if v != None)


def __contains__(self, now):
for k in self.at.keys():
try:
if not getattr(now, k) in self.at[k]: return False
except TypeError:
if self.at[k] != getattr(now, k): return False
return True
小开
最佳答案

如果你正在寻找一些轻量级的东西，签出时间表:

import schedule
import time


def job():
print("I'm working...")


schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)


while 1:
schedule.run_pending()
time.sleep(1)

信息披露:我是那个库的作者。

小开

列出的解决方案甚至都没有尝试解析复杂的cron调度字符串。这是我的版本，使用croniter。基本要点:

schedule = "*/5 * * * *" # Run every five minutes


nextRunTime = getNextCronRunTime(schedule)
while True:
roundedDownTime = roundDownTime()
if (roundedDownTime == nextRunTime):
####################################
### Do your periodic thing here. ###
####################################
nextRunTime = getNextCronRunTime(schedule)
elif (roundedDownTime > nextRunTime):
# We missed an execution. Error. Re initialize.
nextRunTime = getNextCronRunTime(schedule)
sleepTillTopOfNextMinute()

辅助例程:

from croniter import croniter
from datetime import datetime, timedelta


# Round time down to the top of the previous minute
def roundDownTime(dt=None, dateDelta=timedelta(minutes=1)):
roundTo = dateDelta.total_seconds()
if dt == None : dt = datetime.now()
seconds = (dt - dt.min).seconds
rounding = (seconds+roundTo/2) // roundTo * roundTo
return dt + timedelta(0,rounding-seconds,-dt.microsecond)


# Get next run time from now, based on schedule specified by cron string
def getNextCronRunTime(schedule):
return croniter(schedule, datetime.now()).get_next(datetime)


# Sleep till the top of the next minute
def sleepTillTopOfNextMinute():
t = datetime.utcnow()
sleeptime = 60 - (t.second + t.microsecond/1000000.0)
time.sleep(sleeptime)
小开

我知道有很多答案，但另一个解决方案可能是修饰符。这是一个每天在特定时间重复一个函数的例子。使用这种方式的一个很酷的想法是，你只需要将语法糖添加到你想要调度的函数:

@repeatEveryDay(hour=6, minutes=30)
def sayHello(name):
print(f"Hello {name}")


sayHello("Bob") # Now this function will be invoked every day at 6.30 a.m

装饰器看起来像这样:

def repeatEveryDay(hour, minutes=0, seconds=0):
"""
Decorator that will run the decorated function everyday at that hour, minutes and seconds.
:param hour: 0-24
:param minutes: 0-60 (Optional)
:param seconds: 0-60 (Optional)
"""
def decoratorRepeat(func):


@functools.wraps(func)
def wrapperRepeat(*args, **kwargs):


def getLocalTime():
return datetime.datetime.fromtimestamp(time.mktime(time.localtime()))


# Get the datetime of the first function call
td = datetime.timedelta(seconds=15)
if wrapperRepeat.nextSent == None:
now = getLocalTime()
wrapperRepeat.nextSent = datetime.datetime(now.year, now.month, now.day, hour, minutes, seconds)
if wrapperRepeat.nextSent < now:
wrapperRepeat.nextSent += td


# Waiting till next day
while getLocalTime() < wrapperRepeat.nextSent:
time.sleep(1)


# Call the function
func(*args, **kwargs)


# Get the datetime of the next function call
wrapperRepeat.nextSent += td
wrapperRepeat(*args, **kwargs)


wrapperRepeat.nextSent = None
return wrapperRepeat


return decoratorRepeat
小开

我喜欢pycron包解决这个问题的方式。

import pycron
import time


while True:
if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
print('running backup')
time.sleep(60)               # The process should take at least 60 sec
# to avoid running twice in one minute
else:
time.sleep(15)               # Check again in 15 seconds

提交答案