Wie bekomme ich einen Cron-ähnlichen Scheduler in Python?

Question

Ich bin auf der Suche nach einer Bibliothek in Python, die Folgendes bietet at y cron wie Funktionalität.

Ich hätte gerne eine reine Python-Lösung, anstatt mich auf Tools zu verlassen, die auf der Box installiert sind; auf diese Weise kann ich auf Maschinen ohne Cron laufen.

Für diejenigen, die nicht vertraut sind mit cron : Sie können Aufgaben auf der Grundlage eines Ausdrucks wie:

 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.

Die Syntax des Cron-Zeitausdrucks ist weniger wichtig, aber ich würde gerne etwas mit dieser Art von Flexibilität haben.

Wenn es nicht etwas gibt, das dies für mich sofort erledigt, wären wir für jeden Vorschlag für die Bausteine dankbar, um so etwas zu erstellen.

bearbeiten Ich bin nicht daran interessiert, Prozesse zu starten, sondern nur "Jobs", die ebenfalls in Python geschrieben sind - Python-Funktionen. Ich denke, dass dies notwendigerweise ein anderer Thread, aber nicht in einem anderen Prozess sein würde.

Zu diesem Zweck bin ich auf der Suche nach der Ausdruckskraft des Cron-Zeit-Ausdrucks, aber in Python.

Cron a gibt es schon seit Jahren, aber ich versuche, so mobil wie möglich zu sein. Ich kann mich nicht auf seine Anwesenheit verlassen.

Answer 1

Wenn Sie etwas Leichtes suchen, schauen Sie sich die Zeitplan :

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)

Offenlegung : Ich bin der Autor dieser Bibliothek.

Answer 2

Sie können einfach die normale Python-Syntax zur Übergabe von Argumenten verwenden, um Ihre crontab zu spezifizieren. Nehmen wir zum Beispiel an, wir definieren eine Ereignisklasse wie unten:

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)

(Hinweis: Nicht gründlich getestet)

Dann kann Ihr CronTab in normaler Python-Syntax angegeben werden als:

c = CronTab(
  Event(perform_backup, 0, 2, dow=6 ),
  Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)

Auf diese Weise erhalten Sie die volle Leistung der Python-Argument-Mechanik (Mischen von Positions- und Schlüsselwort-Args, und können symbolische Namen für Namen von Wochen und Monaten verwenden)

Die Klasse CronTab würde so definiert werden, dass sie einfach in Minutenschritten schläft und bei jedem Ereignis check() aufruft. (Es gibt wahrscheinlich einige Feinheiten mit Sommerzeit / Zeitzonen vorsichtig sein, aber). Hier ist eine schnelle Implementierung:

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)

Ein paar Dinge zu beachten: Python's Wochentage / Monate sind Null indiziert (im Gegensatz zu cron), und dass Bereich schließt das letzte Element, daher Syntax wie "1-5" wird range(0,5) - dh [0,1,2,3,4]. Wenn Sie die Cron-Syntax bevorzugen, sollte das Parsen jedoch nicht allzu schwierig sein.

Answer 3

Mehr oder weniger dasselbe wie oben, aber gleichzeitig mit 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()

Answer 4

Keine der aufgeführten Lösungen versucht auch nur, eine komplexe Cron-Zeitplan-Zeichenkette zu analysieren. Hier ist also meine Version, die croniter . Das Wesentliche:

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

Helfer-Routinen:

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)

Answer 5

Mir gefällt, wie die pycron Paket löst dieses Problem.

import pycron
import time

while True:
    if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
        print('running backup')
        time.sleep(60)               # The process should take at least 60 sec
                                     # to avoid running twice in one minute
    else:
        time.sleep(15)               # Check again in 15 seconds