Как мне получить Cron-подобный планировщик в Python?

Question

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

Answer 1

Ознакомьтесь с luigi (https://github.com/spotify/luigi).. Он написан на python и имеет приятный веб-интерфейс для мониторинга задач. Он также имеет граф зависимостей. Может быть, излишне для того, что вам нужно, но, вероятно, это поможет.

Answer 2

У меня есть небольшое исправление для метода запуска класса 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()

Answer 3

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

Answer 4

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

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

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)

Answer 5

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

Answer 6

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

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

Answer 7

Я знаю, что ответов много, но другим решением может быть использование декораторов . Это пример повторения функции каждый день в определенное время. Хорошая идея использования этого способа заключается в том, что вам нужно только добавить Синтаксический сахар к функции, которую вы хотите запланировать:

@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

Answer 8

Если вы ищете распределенный планировщик, вы можете проверить https://github.com/sherinkurian/mani - он действительно нуждается в повторном редактировании, хотя, возможно, это не то, что вы ищете. (обратите внимание, что я автор) это было построено для обеспечения отказоустойчивости благодаря работе часов более чем на одном узле.

Answer 9

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

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

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

Answer 10

Если скрипт, который вы хотите запустить, основан на сети, вы можете использовать стороннюю службу, такую ​​как crono , для программной настройки ваших заданий.