understanding script ejecutar con python cron scheduled-tasks

script - ¿Cómo obtengo un programador tipo Cron en Python?



python crontab (20)

Estoy buscando una biblioteca en Python que proporcione una funcionalidad similar a la de cron .

Me gustaría tener una solución Python pura, en lugar de confiar en las herramientas instaladas en la caja; De esta manera corro en máquinas sin cron.

Para aquellos que no están familiarizados con cron : puede programar tareas basadas en una expresión como:

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 sintaxis de la expresión cron time es menos importante, pero me gustaría tener algo con este tipo de flexibilidad.

Si no hay algo que haga esto por mí fuera de la caja, cualquier sugerencia para los componentes básicos para hacer algo como esto sería recibida con gratitud.

Editar No estoy interesado en iniciar procesos, solo "trabajos" también escritos en Python - funciones de python. Por necesidad creo que este sería un hilo diferente, pero no en un proceso diferente.

Para este fin, estoy buscando la expresividad de la expresión del tiempo cron, pero en Python.

Cron ha existido durante años, pero estoy tratando de ser lo más portátil posible. No puedo confiar en su presencia.

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

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

He modificado el guión.

  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

Más o menos igual que el anterior 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()

Ninguna de las soluciones enumeradas intenta siquiera analizar una cadena de programación cron compleja. Entonces, aquí está mi versión, usando croniter . Esencia 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)

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

No sé si ya existe algo así. Sería fácil escribir sus propios módulos de hora, fecha y hora o calendario, visite http://docs.python.org/library/time.html

La única preocupación de 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 sí necesita confiar en las soluciones que dependen del sistema.

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 es:

# 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

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

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

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

Puede usar la sintaxis 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 de Python normal 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, obtienes todo el poder de la mecánica de argumentos de Python (mezclando argumentos posicionales y de palabras clave, y puedes usar nombres simbólicos para nombres de semanas y meses)

La clase CronTab se definiría como simplemente dormir en incrementos de minutos y llamar a check () en cada evento. (Probablemente hay algunas sutilezas con el horario de verano o las zonas horarias de las que hay que 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 / mes de Python están indexados en cero (a diferencia de cron), y ese rango excluye el último elemento, por lo que la sintaxis como "1-5" se convierte en rango (0,5), es decir, [0,1,2, 3,4]. Si prefieres la sintaxis cron, el análisis no debería ser demasiado difícil.

Si está buscando un programador distribuido, puede consultar https://github.com/sherinkurian/mani , aunque necesita un redis, por lo que puede que no sea lo que está buscando. (tenga en cuenta que soy el autor) esto se creó para garantizar la tolerancia a fallos al hacer que el reloj se ejecute en más de un nodo.

Si estás buscando un schedule pago ligero:

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)

Divulgación : Soy el autor de esa biblioteca.

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

El tiempo se agotó en 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()

Una cosa que en mis búsquedas he visto es el módulo de programación de python, que podría ser el tipo de cosa que estás buscando.

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)

TurboGears envía con capacidad de tareas programadas 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.