Como consigo um cron como scheduler em Python?

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/

Vejamo aipo , eles têm tarefas periódicas como o cron.

"... 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.

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

Uma coisa que vi nas minhas pesquisas foi a do python.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.

Eu modifiquei o guião.

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. Tente iniciar a tarefa no primeiro segundo de um minuto.

Código no Github

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

Olha o luigi. https://github.com/spotify/luigi . está escrito em python e tem uma boa interface web para tarefas de monitorização. Ele também tem um gráfico de dependência. Pode ser exagero para o que você precisa, mas provavelmente vai fazer o truque.

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.

A solução do Brian está a funcionar muito bem. No entanto, como outros já apontaram, há um bug sutil no código de execução. Também achei demasiado complicado para as necessidades.

Aqui está a minha alternativa mais simples e funcional para o código de execução, caso alguém precise dele.

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ó.

Não sei se algo assim já existe. Seria fácil escrever o seu próprio com módulos de tempo, datetime e/ou calendário, Ver http://docs.python.org/library/time.html

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.

Eu peguei na solução do Brian, fiz algumas alterações, adicionei o início de um analisador de arquivos padrão crontab, e coloquei em {[[2]}https://bitbucket.org/dbenamy/devcron.

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.

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

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