Como consigo um cron como scheduler em Python?
Estou à procura de uma biblioteca em Python que irá fornecer a funcionalidade at
e cron
como a funcionalidade.
Eu gostaria de ter uma solução Python pura, em vez de confiar em ferramentas instaladas na caixa; desta forma Eu corro em máquinas sem cron.
para aqueles que não estão familiarizados com cron
: pode agendar tarefas com base numa expressão 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.
a sintaxe de expressão de tempo de cron é menos importante, mas eu gostaria de ter algo com este tipo de flexibilidade.
Se não houver algo que faça isto por mim fora da caixa, qualquer sugestão para os blocos de construção fazerem algo assim seria recebida com gratidão.editar Não estou interessado em lançar processos, apenas "empregos" também escritos em funções Python - python. Por necessidade, penso que este seria um fio diferente, mas não num processo diferente.
Para este fim, procuro a expressividade da expressão do tempo de cron, mas em Python.Cronexiste há anos, mas estou a tentar ser o mais portátil possível. Não posso confiar na sua presença.
20 answers
Se está à procura de algo leve check-out escalonamento:
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)
sou o autor dessa biblioteca.
Pode apenas usar a sintaxe normal de passagem do argumento Python para indicar o seu crontab. Por exemplo, suponha que definimos uma classe de eventos como abaixo:
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: não completamente testado)
Então o seu CronTab pode ser especificado na sintaxe normal em python como:
c = CronTab(
Event(perform_backup, 0, 2, dow=6 ),
Event(purge_temps, 0, range(9,18,2), dow=range(0,5))
)
Desta forma você obtém todo o poder da mecânica dos argumentos de Python (misturando posicionais e palavras-chave args, e pode usar nomes simbólicos para nomes de semanas e meses)
A classe CronTab seria definida como simplesmente dormindo em incrementos de minutos, e chamando check() em cada evento. (Há provavelmente algumas sutilezas com horário de Verão / fusos horários a serem cautelosos). Aqui está uma implementação 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)
Algumas coisas a notar: os dias úteis / meses do Python são indexados zero (ao contrário do cron), e esse intervalo exclui o último elemento, daí a sintaxe como "1-5" torna - se range(0,5) - ie [0,1,2,3,4]. Se você preferir a sintaxe de cron, analisá-la não deve ser muito difícil no entanto.
Talvez isto tenha surgido apenas depois da pergunta ter sido feita; pensei apenas mencioná-lo por uma questão de exaustividade: https://apscheduler.readthedocs.org/en/latest/
"... Módulo Crontab para ler e escrever arquivos crontab e acessar o sistema cron automaticamente e simplesmente usando uma API direta. ..."
Http://pypi.python.org/pypi/python-crontab
E também APScheduler, um pacote python. Já foi escrita e depurada.
sched
módulo que pode ser o tipo de coisa que procuras.
Mais ou menos igual ao acima, mas em simultâneo com 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 Navios com capacidade de tarefa programada baseada em Kronos
Nunca usei o Kronos directamente, mas a programação no TG tem um conjunto decente de características e é sólida.-
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()
Tente iniciar a tarefa no primeiro segundo de um minuto.
Nenhuma das soluções listadas sequer tenta processar um complexo cronograma cron string. Então, aqui está a minha versão, usando croniter . Gist básico:
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()
Rotinas auxiliares:
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)
Tenho uma pequena correcção para o método de execução da classe CronTab sugerido pelo Brian .
O tempo foi ultrapassado por um segundo, levando a um ciclo de um segundo, difícil no 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()
Não existe uma forma" python puro " de fazer isto porque algum outro processo teria de lançar python para executar a sua solução. Cada plataforma terá uma ou vinte maneiras diferentes de lançar processos e monitorar seu progresso. Em plataformas unix, cron é o padrão antigo. No Mac OS X há também o launchd, que combina lançamento tipo cron com funcionalidade watchdog que pode manter o seu processo vivo, se é isso que você quer. Uma vez que python está em execução, então você pode usar o Módulo sched para agendar tarefas.
Só para o caso de, se estiver a usar o windows, existir um pycron. Check out http://sourceforge.net/projects/pycron / . Para o linux, vou chamar-me cron ou sched.
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)
Outra solução trivial seria:
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)
E a classe aqcron.At riz:
# 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
Se está à procura de um escalonador distribuído, pode verificar https://github.com/sherinkurian/mani mas precisa de redis, por isso pode não ser o que procura. (note que eu sou o autor) isto foi construído para garantir a tolerância a falhas por ter o relógio rodado em mais de um nó.
A única preocupação para uma solução em python é que o seu trabalho precisa estar sempre em execução e possivelmente ser automaticamente "ressuscitado" após um reboot, algo para o qual você faz precisa confiar em soluções dependentes do sistema.
Você pode verificar os [1] Crons de PiCloud [2], mas lembre-se que seus trabalhos não estarão funcionando em sua própria máquina. É também um serviço que você precisará pagar se você usar mais de 20 horas de tempo de cálculo por mês.