¿Cómo obtengo un Cron como scheduler en Python?

Simplemente podría usar la sintaxis normal de paso de argumentos de Python para especificar su crontab. Por ejemplo, supongamos que definimos una clase de evento de la siguiente manera:

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)

(Nota: No probado a fondo)

Entonces su CronTab se puede especificar en la sintaxis normal de python como:

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

De esta manera se obtiene toda la potencia de la mecánica argumental de Python (mezclando argumentos posicionales y de palabras clave, y se pueden usar nombres simbólicos para nombres de semanas y meses)

La clase CronTab se definiría como simplemente durmiendo en incrementos de minutos, y llamando a check () en cada evento. (Probablemente hay algunas sutilezas con el horario de verano / zonas horarias a tener cuidado). Aquí hay una implementación rápida:

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)

Algunas cosas a tener en cuenta: Los días de la semana / meses de Python están indexados a cero (a diferencia de cron), y ese rango excluye el último elemento, por lo tanto una sintaxis como "1-5" se convierte en rango(0,5) - ie [0,1,2,3,4]. Si prefiere la sintaxis cron, sin embargo, analizarla no debería ser demasiado difícil.

Tal vez esto ha surgido solo después de que se hizo la pregunta; pensé que solo lo mencionaba para completar: https://apscheduler.readthedocs.org/en/latest /

Echa un vistazo Apio, tienen tareas periódicas como cron.

"... Módulo Crontab para leer y escribir archivos crontab y acceder al cron del sistema automáticamente y simplemente usando una API directa. ..."

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

Y también APScheduler, un paquete python. Ya está escrito y depurado.

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

Una cosa que he visto en mis búsquedas es la de Python sched módulo que podría ser el tipo de cosa que estás buscando.

Más o menos igual que arriba pero concurrente usando 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()

TurboGears se envía con capacidad de tarea programada basada en Kronos

Nunca he usado Kronos directamente, pero la programación en TG tiene un conjunto decente de características y es sólida.

He modificado el script.

1. Fácil de usar:

   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. Trate de iniciar la tarea en el primer segundo de un minuto.

Código en Github

Ninguna de las soluciones listadas intenta siquiera analizar una cadena de programación cron compleja. Entonces, aquí está mi versión, usando croniter . Síntesis básica:

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

Rutinas de ayuda:

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)

Tengo una corrección menor para el método de ejecución de la clase CronTab sugerido por Brian.

El tiempo se agotó por un segundo, lo que llevó a un bucle duro de un segundo al final de cada minuto.

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

Echa un vistazo a luigi ( https://github.com/spotify/luigi ). Está escrito en python y tiene una interfaz de usuario web agradable para tareas de monitoreo. También tiene un gráfico de dependencias. Podría ser exagerado para lo que necesita, pero probablemente hará el truco.

No hay una forma "pura de python" de hacer esto porque algún otro proceso tendría que lanzar python para ejecutar su solución. Cada plataforma tendrá una o veinte formas diferentes de lanzar procesos y monitorear su progreso. En plataformas unix, cron es el viejo estándar. En Mac OS X también está launchd, que combina el lanzamiento similar a cron con la funcionalidad watchdog que puede mantener vivo su proceso si eso es lo que desea. Una vez que python se está ejecutando, entonces puede utilizar el sched módulo para programar tareas.

Por si acaso, si estás usando windows, existe un pycron. Echa un vistazo http://sourceforge.net/projects/pycron/ . Para Linux, voy a ir por cron o sched.

La solución de Brian está funcionando bastante bien. Sin embargo, como otros han señalado, hay un error sutil en el código de ejecución. También me pareció demasiado complicado para las necesidades.

Aquí está mi alternativa más simple y funcional para el código de ejecución en caso de que alguien lo necesite:

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)

Otra solución trivial sería:

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)

Y la clase aqcron.At is:

# 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

Si está buscando un programador distribuido, puede consultar https://github.com/sherinkurian/mani - aunque sí necesita redis, puede que no sea lo que estás buscando. (tenga en cuenta que soy el autor) esto fue construido para garantizar la tolerancia a errores al tener reloj ejecutado en más de un nodo.

No se si algo así ya existe. Sería fácil escribir el suyo propio con módulos time, datetime y/o calendar, ver http://docs.python.org/library/time.html

La única preocupación para una solución de python es que su trabajo debe estar siempre en ejecución y posiblemente ser "resucitado" automáticamente después de un reinicio, algo para lo que hace necesita confiar en soluciones dependientes del sistema.

Tomé la solución de Brian, hice algunos cambios, agregué los inicios de un analizador de archivos crontab estándar, y lo puse en https://bitbucket.org/dbenamy/devcron .

Puede consultar los Crones de PiCloud [1] [2], pero tenga en cuenta que sus trabajos no se ejecutarán en su propia máquina. También es un servicio que tendrá que pagar si utiliza más de 20 horas de tiempo de cómputo al mes.

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

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