¿Cómo obtengo un programador tipo Cron en Python?
Estoy buscando una biblioteca en Python que proporcione atuna cronfuncionalidad similar.
Me gustaría bastante tener una solución Python pura, en lugar de depender de herramientas instaladas en la caja; De esta manera ejecuto en máquinas sin cron.
Para aquellos que no están familiarizados con cron: puede programar tareas basándose 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 de tiempo cron es menos importante, pero me gustaría tener algo con este tipo de flexibilidad.
Si no hay algo que haga esto por mí de inmediato, cualquier sugerencia sobre 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 ello busco la expresividad de la expresión de tiempo cron, pero en Python.
Cron existe desde hace años, pero estoy tratando de ser lo más portátil posible. No puedo confiar en su presencia.
Si buscas algo ligero, programa de pago :
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.
Podrías usar la sintaxis normal de paso de argumentos de Python para especificar tu 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 exhaustivamente)
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, obtiene todo el poder de la mecánica de argumentos de Python (mezcla argumentos posicionales y de palabras clave, y puede usar nombres simbólicos para nombres de semanas y meses)
La clase CronTab se definiría simplemente durmiendo en incrementos de minutos y llamando a check() en cada evento. (Sin embargo, probablemente haya algunas sutilezas con el horario de verano/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/meses de la semana de Python están indexados a cero (a diferencia de cron), y ese rango excluye el último elemento, por lo tanto, la sintaxis como "1-5" se convierte en rango(0,5), es decir, [0,1,2, 3,4]. Sin embargo, si prefiere la sintaxis cron, analizarla no debería ser demasiado difícil.
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()
