如何在Python中获得像调度程序的Cron？
我正在寻找一个Python的库，它将提供像和cron一样的function。
我很想拥有一个纯粹的Python解决scheme，而不是依靠安装在盒子上的工具。 这样我在没有cron的机器上运行。
对于那些不熟悉cron人来说：你可以根据如下expression式来安排任务：
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时间expression式语法不那么重要，但是我想要具有这种灵活性。
如果对我来说没有什么特别的东西，那么任何关于构build模块的build议都会受到感谢。
编辑我对启动进程不感兴趣，只是“作业”也用Python写的 – python函数。 根据需要，我认为这将是一个不同的线程，但不是在一个不同的过程。
为此，我正在寻找cron时间expression式的performance力，但在Python中。
克朗已经存在多年了，但我正在尽可能地携带。 我不能依靠它的存在。
如果你正在寻找一些轻量级结账时间表 ：
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)
披露 ：我是该图书馆的作者。
你可以使用普通的Pythonparameter passing语法来指定你的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)
（注意：未经彻底testing）
那么你的CronTab可以用普通的Python语法来指定：
c = CronTab( Event(perform_backup, 0, 2, dow=6 ), Event(purge_temps, 0, range(9,18,2), dow=range(0,5)) )
通过这种方式，您可以获得Python参数机制的全部function（混合位置参数和关键字参数，并且可以使用符号名称来表示数周和数月的名称）
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） – 即[0,1,2， 3,4]。 如果你喜欢cron语法，parsing它应该不是太困难。
也许这个问题是在问题出现之后才提出的; 我想我只是完整地提到它： https ： //apscheduler.readthedocs.org/en/latest/
看看芹菜 ，他们有周期性的任务像cron。
“…用于读取和写入crontab文件并自动访问系统cron的Crontab模块，只需使用直接的API …”
http://pypi.python.org/pypi/python-crontab
还有一个python包APScheduler。 已经编写和debugging。
在我的search中，我所看到的一件事是python的sched模块，这可能是你正在寻找的东西。
TurboGears提供基于Kronos的计划任务function
我从来没有直接使用Kronos，但TG中的调度有一个体面的function，是坚实的。
如上所述或多或less相同，但同时使用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()
我修改了脚本。
-
使用方便：
cron = Cron() cron.add('* * * * *' , minute_task) # every minute cron.add('33 * * * *' , day_task) # every hour cron.add('34 18 * * *' , day_task) # every day cron.run() -
尝试在一分钟的头一秒开始任务。
Github上的代码
我对Brianbuild议的CronTab类运行方法有一个小小的修正。
时机已经过去了一秒，导致每分钟结束时只有一秒的硬循环。
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) n = datetime.now() while n < t: s = (t - n).seconds + 1 time.sleep(s) n = datetime.now()
看看luigi（ https://github.com/spotify/luigi ）。 它是用python编写的，并有一个很好的Web UI用于监视任务。 它也有一个依赖图。 可能是为了你需要的矫枉过正，但它可能会诀窍。
列出的解决scheme甚至没有尝试parsing一个复杂的cron调度string。 所以，这里是我的版本，使用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)
没有“纯python”的方式来做到这一点，因为其他进程将不得不启动python才能运行你的解决scheme。 每个平台将有一到两种不同的方式来启动stream程并监控其进展。 在unix平台上，cron是旧标准。 在Mac OS X上还有launchd，它将cron-like启动与看门狗function相结合，如果这是你想要的，可以保持你的进程活着。 一旦python运行，那么你可以使用调度模块来安排任务。
以防万一，如果你正在使用Windows，存在一个pycron。 看看http://sourceforge.net/projects/pycron/ 。 对于Linux，我会去cron或sched。
另一个微不足道的解决scheme是：
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.At是：
# 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
我不知道这样的事情是否已经存在。 使用时间，date时间和/或日历模块编写自己的代码很容易，请参阅http://docs.python.org/library/time.html
python解决scheme唯一的担心就是你的工作需要一直运行，并且在重新启动之后可能会自动“复活”，对此你需要依靠依赖系统的解决scheme。
Brian的解决scheme工作得很好。 但是，正如其他人所指出的，运行代码中存在一个微妙的错误。 另外我发现它太复杂的需求。
这里是我的运行代码的情况下，任何人都需要它的更简单和function的select：
def run(self): while 1: t = datetime.now() for e in self.events: e.check(t) time.sleep(60 - t.second - t.microsecond / 1000000.0)
我采取了布赖恩的解决scheme，做了一些改变，添加了一个标准的crontab文件parsing器的开始，并把它放在https://bitbucket.org/dbenamy/devcron 。
您可以查看PiCloud的[1] Crons [2]，但请注意您的作业不会在您自己的机器上运行。 如果您每月使用超过20个小时的计算时间，那么您还需要付费。
Jython的？
我刚刚在这里search，发现这个词在这个页面上没有提到。 真正的multithreading（不是多进程），Futures，ExecutorService以及并发工具箱中的所有其他工具等等。
