Как мне получить Cron-подобный планировщик в Python?
Я ищу библиотеку на 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, но в Python.
Cron существует уже много лет, но я стараюсь быть максимально переносимым. Я не могу полагаться на его наличие.
Переведено автоматически
Ответ 1
Если вы ищете что-то облегченное, оформите расписание:
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)
Раскрытие информации: я автор этой библиотеки.
Ответ 2
Вы могли бы просто использовать обычный синтаксис передачи аргументов Python, чтобы указать свой crontab. Например, предположим, что мы определяем класс событий, как показано ниже:
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) - т.е. [0,1,2,3,4]. Однако, если вы предпочитаете синтаксис cron, его синтаксический анализ не должен быть слишком сложным.
Ответ 3
Более или менее такой же, как указано выше, но с одновременным использованием 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()
Ответ 4
Croniter: анализирует строку CRON и получает последовательность следующих событий
Ни одно из перечисленных решений не использует подход синтаксического анализа строки расписания CRON. Итак, вот моя версия, использующая croniter. Основная суть:
import croniter
import datetime
now = datetime.datetime.now()
iter = croniter.croniter('0 0 * * *', now)
next_datetime = croniter.get_next(datetime.datetime) # datetime not now!
remaining_seconds = next_datetime - now
remaining_seconds = remaining_seconds.total_seconds()
print(f'seconds until next event: {remaining_seconds}')
Просто!
Более сложные варианты использования с более подробной логикой:
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)