Как получить планировщик Cron в 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) 

Раскрытие информации : Я являюсь автором этой библиотеки.

Вы можете просто использовать обычный синтаксис передачи аргументов 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 (смешивая позиционные и ключевые слова args и можете использовать символические имена для имен недель и месяцев)

Класс 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, разбор его не должен быть слишком сложным.

возможно, это возникло только после того, как был задан вопрос; Я думал, что просто говорю об этом ради полноты: https://apscheduler.readthedocs.org/en/latest/

Посмотрите на сельдерей , у них есть периодические задания, такие как cron.

«… Модуль Crontab для чтения и записи файлов crontab и доступа к системе cron автоматически и просто с использованием прямого API …»

http://pypi.python.org/pypi/python-crontab

а также APScheduler, пакет python. Уже написано и отлажено.

http://packages.python.org/APScheduler/cronschedule.html

Одна вещь, которую я видел в моих поисках, – это модуль sched python, который может быть тем, что вы ищете.

TurboGears поставляется с возможностью запланированной задачи, основанной на Kronos

Я никогда не использовал Kronos напрямую, но планирование в TG имеет приличный набор функций и является прочным.

Более или менее то же, что и выше, но параллельно с использованием 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() 

Я изменил сценарий.

1. Легко использовать:

    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() 

2. Попробуйте запустить задачу за первую секунду.

Код на Github

У меня есть небольшое исправление для метода запуска класса 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 и имеет хороший веб-интерфейс для мониторинга задач. Он также имеет граф зависимостей. Возможно, это будет излишним для того, что вам нужно, но это, вероятно, сделает трюк.

Ни одно из перечисленных решений даже не пытается проанализировать сложную строку расписания 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) 

Для этого не существует «чистого питона», потому что для запуска другого решения потребуется запустить python. На каждой платформе будет один или два разных способа запуска процессов и мониторинг их прогресса. На платформах unix cron является старым стандартом. В Mac OS X также есть launchd, который сочетает в себе cron-подобный запуск с функциями сторожевого таймера, которые могут поддерживать ваш процесс, если вы этого хотите. После запуска python вы можете использовать модуль расписания для планирования задач.

На всякий случай, если вы используете окна, существует pycron. Проверьте http://sourceforge.net/projects/pycron/ . Для linux я перейду в cron или sched.

Другое тривиальное решение:

 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 

Я не знаю, существует ли что-то подобное. Было бы легко написать свои собственные с помощью модулей времени, времени и / или календаря, см. http://docs.python.org/library/time.html

Единственная проблема для решения python заключается в том, что ваша работа должна быть всегда запущена и, возможно, автоматически «воскрешена» после перезагрузки, для чего вам нужно опираться на системно-зависимые решения.

Решение Брайана работает достаточно хорошо. Однако, как указывали другие, в коде запуска есть тонкая ошибка. Также я нашел это слишком сложным для нужд.

Вот моя простая и функциональная альтернатива для кода запуска, если кому-то это понадобится:

 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) 

Я принял решение Брайана, внес несколько изменений, добавил начало стандартного парсера файла crontab и поместил его на https://bitbucket.org/dbenamy/devcron .

Вы можете проверить [1] Crons [1] PiCloud, но обратите внимание, что ваши задания не будут выполняться на вашей собственной машине. Это также услуга, которую вам нужно будет заплатить, если вы используете более 20 часов вычислительного времени в месяц.

[1] http://www.picloud.com

[2] http://docs.picloud.com/cron.html

Jython?

Я просто искал здесь и обнаружил, что это слово не было упомянуто на этой странице. Реальная многопоточность (не многопроцессорная), Futures, ExecutorService и все другие инструменты в панели инструментов параллелизма и т. Д.