Comment puis-je obtenir un cron comme scheduler en Python?

je suis à la recherche d'une bibliothèque en Python qui fournira at et cron comme fonctionnalité.

j'aimerais bien avoir une solution Python pure, plutôt que de compter sur des outils installés sur la boîte; de cette façon je tourne sur des machines sans cron.

pour ceux qui ne connaissent pas cron : vous pouvez programmer des tâches basées sur une expression comme:

 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.

la syntaxe d'expression de temps cron est moins important, mais je voudrais avoir quelque chose avec ce genre de flexibilité.

S'il n'y a pas quelque chose qui fait cela pour moi, toutes les suggestions pour les blocs de construction pour faire quelque chose comme ceci seraient reçues avec gratitude.

Modifier Je ne suis pas intéressé par les processus de lancement, juste les "jobs" aussi écrits dans les fonctions Python - python. Par nécessité, je pense que ce serait un fil différent, mais pas dans un autre processus.

à cette fin, je cherche l'expressivité de l'expression temporelle cron, mais en Python.

Cron a été autour depuis des années, mais je vais essayer d'être aussi portable que possible. Je ne peut pas compter sur sa présence.

249
demandé sur Damjan Pavlica 2008-12-17 03:56:31

20 réponses

si vous cherchez quelque chose de léger caisse horaire :

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)

Divulgation : je suis l'auteur de la bibliothèque.

369
répondu dbader 2016-04-23 21:46:55

vous pouvez simplement utiliser la syntaxe de passage d'argument Python normale pour spécifier votre crontab. Par exemple, supposons que nous définissions une classe D'événement comme suit:

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)

(Note: non testé à fond)

alors votre CronTab peut être spécifié dans la syntaxe normale de python comme:

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

de cette façon, vous obtenez la pleine puissance de la mécanique d'argument de Python (mélange de position et de mots clés args, et peut utiliser des noms symboliques pour les noms de semaines et mois)

la classe CronTab se définirait simplement comme dormant par incréments de minutes, et appelant check() sur chaque événement. (Il ya probablement quelques subtilités avec l'heure d'été / fuseaux horaires à se méfier de cependant). Voici une implémentation rapide:

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)

quelques choses à noter: les jours / mois de Python sont indexés à zéro (contrairement à cron), et cette plage exclut le dernier élément, donc la syntaxe comme " 1-5 " devient plage (0,5) - ie [0,1,2,3,4]. Si vous préférez la syntaxe cron, l'analyse ne devrait pas être trop difficile.

58
répondu Brian 2009-05-11 17:09:15

peut-être que c'est arrivé seulement après que la question ait été posée; j'ai pensé que je l'ai juste mentionné pour être complet: https://apscheduler.readthedocs.org/en/latest/

39
répondu ssc 2014-02-07 02:08:01

Check out Celery , ils ont des tâches périodiques comme cron.

27
répondu Vishal 2014-09-02 10:59:40

"... Module Crontab pour lire et écrire des fichiers crontab et accéder au système cron automatiquement et simplement en utilisant une API directe. ..."

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

et aussi APScheduler, un paquet python. Déjà écrit et débogué.

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

18
répondu bootload 2012-12-12 02:51:19

une chose que j'ai vue dans Mes recherches est le module sched de python qui pourrait être le genre de chose que vous recherchez.

14
répondu Sean 2008-12-17 01:45:17

plus ou moins le même que ci-dessus mais en utilisant simultanément 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()
10
répondu Hackeron 2010-06-01 02:01:21

TurboGears navires ayant une capacité opérationnelle prévue basée sur Kronos

Je n'ai jamais utilisé Kronos directement, mais la programmation dans TG a un ensemble décent de caractéristiques et est solide.

9
répondu James Brady 2008-12-17 06:23:28

j'ai modifié le script.

  1. facile à utiliser:

    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. Essayez de démarrer la tâche dans la première seconde d'une minute.

Code sur Github

7
répondu ning 2014-08-02 20:54:31

aucune des solutions énumérées ne tente même d'analyser une chaîne de programme cron complexe. Donc, voici ma version, en utilisant croniter . Base résumé:

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

routines D'aide:

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
répondu rouble 2018-07-03 15:42:16

j'ai une correction mineure pour la méthode D'exécution de classe crontab suggérée par Brian .

Le timing était par une seconde conduisant à une seconde, dur boucle à la fin de chaque minute.

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()
6
répondu benc 2017-05-23 12:18:24

découvrez luigi ( https://github.com/spotify/luigi ). Il est écrit en python et dispose d'une interface Web agréable pour surveiller les tâches. Il dispose également d'un graphe de dépendance. Peut-être trop pour ce dont vous avez besoin, mais ça fera probablement l'affaire.

5
répondu amwinter 2013-09-30 16:57:26

il n'y a pas de" Python pur " pour faire cela parce qu'un autre processus devrait lancer python pour exécuter votre solution. Chaque plate-forme aura une ou vingt façons différentes de lancer des processus et de suivre leurs progrès. Sur les plateformes unix, cron est l'ancienne norme. Sur Mac OS X il y a aussi launchd, qui combine le lancement de type cron avec une fonctionnalité de chien de garde qui peut garder votre processus vivant si c'est ce que vous voulez. Une fois que python est en cours d'exécution, vous pouvez utiliser le module sched pour programmer les tâches.

3
répondu Nick 2008-12-17 05:45:01

Juste au cas où, si vous utilisez windows, il existe un pycron. Vérifier http://sourceforge.net/projects/pycron / . Pour linux, je vais passer par cron ou sched.

1
répondu JV. 2008-12-17 05:49:29

la solution de Brian fonctionne plutôt bien. Cependant, comme d'autres l'ont souligné, il existe une subtile bug dans le code d'exécution. Aussi je l'ai trouvé trop compliqué pour les besoins.

Voici mon alternative plus simple et fonctionnelle pour le code d'exécution au cas où quelqu'un en aurait besoin:

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)
1
répondu raph.amiard 2017-05-23 11:47:26

une autre solution insignifiante serait:

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)

et la classe aqcron.At est:

# 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
1
répondu fdb 2012-09-10 22:51:04

si vous recherchez un agenda distribué, vous pouvez cocher la case https://github.com/sherinkurian/mani - il a besoin de redis bien que donc pourrait ne pas être ce que vous recherchez. (notez que je suis l'auteur) ceci a été construit pour assurer la tolérance de défaut en ayant l'horloge tourne sur plus d'un noeud.

1
répondu shrnkrn 2018-05-19 16:53:01

Je ne sais pas si quelque chose comme ça existe déjà. Il serait facile d'écrire les vôtres avec des modules time, datetime et/ou calendar, voir http://docs.python.org/library/time.html

le seul souci pour une solution python est que votre travail doit être toujours en cours d'exécution et peut-être être automatiquement" ressuscité "après un redémarrage, quelque chose pour lequel vous do besoin de compter sur des solutions dépendantes du système.

0
répondu Davide 2008-12-17 01:01:33

J'ai pris la solution de Brian, j'ai fait quelques changements, j'ai ajouté les débuts d'un analyseur de fichier crontab standard, et je l'ai mis à https://bitbucket.org/dbenamy/devcron .

0
répondu Dan Benamy 2010-12-24 10:43:54

vous pouvez consulter les [1] Crons de PiCloud [2], mais notez que vos travaux ne seront pas exécutés sur votre propre machine. C'est aussi un service que vous devrez payer si vous utilisez plus de 20 heures de temps de calcul d'un mois.

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

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

0
répondu BrainCore 2013-04-25 00:23:47