Как мне получить Cron-подобный планировщик в Python? - PullRequest
284 голосов
/ 17 декабря 2008

Я ищу библиотеку на 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 существует уже много лет, но я стараюсь быть максимально портативным. Я не могу полагаться на его присутствие.

Ответы [ 23 ]

447 голосов
/ 28 мая 2013

Если вы ищете что-нибудь легкое оформление заказа расписание :

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)

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

61 голосов
/ 17 декабря 2008

Вы можете просто использовать обычный синтаксис передачи аргументов 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», становится range (0,5) - то есть [0,1 2,3,4]. Если вы предпочитаете синтаксис cron, его синтаксический анализ не должен быть слишком сложным.

45 голосов
/ 10 марта 2011

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

28 голосов
/ 20 июля 2010

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

19 голосов
/ 12 декабря 2012

"... Модуль Crontab для чтения и записи файлов crontab и автоматического доступа к системному cron с помощью прямого API. ..."

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

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

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

16 голосов
/ 17 декабря 2008

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

10 голосов
/ 01 июня 2010

Более или менее такой же, как указано выше, но одновременно с использованием 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()
9 голосов
/ 17 декабря 2008

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

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

9 голосов
/ 07 марта 2016

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

Я изменил скрипт.

  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

...