Como obtenho um Cron como agendador em Python? [fechadas]

348

Estou procurando uma biblioteca em Python que forneça ate crongere funcionalidade.

Eu gostaria de ter uma solução Python pura, em vez de confiar nas ferramentas instaladas na caixa; desta forma eu corro em máquinas sem cron.

Para aqueles que não estão familiarizados com cron: você pode agendar tarefas com base em uma 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 da expressão cron time é menos importante, mas eu gostaria de ter algo com esse tipo de flexibilidade.

Se não houver algo que faça isso imediatamente, todas as sugestões para os blocos de construção para fazer algo assim serão recebidas com gratidão.

Editar Não estou interessado em iniciar processos, apenas "trabalhos" também escritos nas funções Python - python. Por necessidade, acho que esse seria um segmento diferente, mas não em um processo diferente.

Para esse fim, procuro a expressividade da expressão cron time, mas em Python.

Cron existe há anos, mas estou tentando ser o mais portátil possível. Não posso confiar na presença dele.

jamesh
fonte
11
Eu também gostaria de saber como fazer isso. Seria mais útil ter uma solução de plataforma cruzada do que depender de componentes específicos da plataforma.
217 Sean
6
Isso não é fora de tópico, é uma pergunta muito importante e útil #
Connor

Respostas:

571

Se você estiver procurando por algo leve na agenda de checkout :

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)

Divulgação : sou o autor dessa biblioteca.

dbader
fonte
7
Você deve mencionar que é o mantenedor de schedule. Isso funcionou bem para mim. Seria ainda melhor se tivesse cron como sintaxe e decoradores suportados (consulte o crython, mas não use esta biblioteca porque não funciona; o agendamento não parece estar bem escrito).
precisa saber é o seguinte
23
Existe uma maneira de passar um parâmetro para o trabalho? Eu gostaria de fazer algo parecido com isto:. Schedule.every () hour.do (trabalho (myParam))
Zen Skunkworx
5
schedule.every (). hour.do (job) isso é executado a cada hora do relógio? Como 01:00, 02:00, 03:00, etc.? mesmo que o horário de início não seja uma hora inteira?
swateek
11
suponha que esse código esteja em scheduler.py. esse código será executado automaticamente?
Kishan
25
@ darrel-holt e @ zen-skunkworx: a do()função encaminha argumentos extras que você passa para a função de trabalho: schedule.readthedocs.io/en/stable/api.html#schedule.Job.do Por exemplo, você pode fazer isso : schedule.every().hour.do(job, param1, param2)Não há necessidade de usar um lambda. Espero que isso ajude :)
dbader
65

Você pode usar o argumento normal do Python passando a sintaxe para especificar seu crontab. Por exemplo, suponha que definimos uma classe Event 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 testado completamente)

Em seguida, seu CronTab pode ser especificado na sintaxe python normal como:

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

Dessa forma, você obtém todo o poder da mecânica de argumentos do Python (misturando argumentos posicionais e de palavras-chave e pode usar nomes simbólicos para nomes de semanas e meses)

A classe CronTab seria definida como simplesmente dormir em incrementos de minutos e chamar check () em cada evento. (Provavelmente existem algumas sutilezas com o horário de verão / fusos horários para se ter cuidado). 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 serem observadas: os dias da semana / meses do Python são zero indexados (diferente do cron), e esse intervalo exclui o último elemento; portanto, sintaxe como "1-5" se torna intervalo (0,5) - ou seja, [0,1,2, 3,4] Se você preferir a sintaxe do cron, a análise não deve ser muito difícil.

Brian
fonte
Você pode querer adicionar algumas instruções de importação para os inexperientes. Eu acabei colocando todas as classes em um único arquivo com from datetime import * from time import sleep e alterei time.sleep para sleep. Solução agradável e simples e elegante. Obrigado.
Mavnn
11
Imaginando, por que isso está sendo preferido em relação a Cronos? O sched é buggy (já que o kronos usa o sched)? Ou isso está desatualizado?
precisa saber é o seguinte
Obrigado Brian, eu uso sua solução em produção e está funcionando muito bem. No entanto, como outros já apontaram, há um bug sutil no seu código de execução. Também achei excessivamente complicado para as necessidades.
Raph.amiard
11
Isso é legal, mas ainda não suporta notação de barra, para execução a cada hora, minuto, etc ...
Chris Koston
11
Excelente idéia para escrever suas próprias classes, por exemplo, quando eu não tiver acesso sudo em um servidor e, portanto, não pode pip install anything:)
Cometsong
46

talvez isso tenha surgido somente depois que a pergunta foi feita; Eu pensei em mencioná-lo por uma questão de integridade: https://apscheduler.readthedocs.org/en/latest/

ssc
fonte
O Apscheduler é extremamente funcional e fácil de usar.
Hatshepsut
27

Uma coisa que nas minhas pesquisas que vi é o schedmódulo python, que pode ser o tipo de coisa que você está procurando.

Sean
fonte
11
sched agora possui enterabs (), que faz o planejamento absoluto.
precisa saber é o seguinte
5
Eu esperaria que essa fosse a resposta preferida, pois o sched faz parte do python2 e 3 stdlib agora e faz o planejamento absoluto.
Michael Michael
11

Mais ou menos o mesmo que acima, mas simultaneamente 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()
Hackeron
fonte
Apenas uma nota que datetime.timetuple () começará com o ano, mês, dia ... etc ...
Trey Stout
9

Nenhuma das soluções listadas sequer tenta analisar uma sequência complexa de cronograma cron. Então, aqui está a minha versão, usando croniter . Essência 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()

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)
rublo
fonte
Como alguém poderia entrar no "perdeu uma execução" elif? Atm Estou usando uma programação como essa "* * * * *", adicionando time.sleepmais de 1 minuto na seção "Faça sua coisa periódica" if, mas sempre vejo as coisas nessa instrução if. Quando leva mais de 1 minuto, vejo o loop while pulando a execução do loop ausente.
TPPZ 02/07/19
@TPPZ O processo pode ter sido suspenso, o relógio pode ter sido alterado manualmente ou por ntp etc. etc. O Croniter é usado no Airflow e parece ter mais recursos do que o módulo Crontab e outros.
dlamblin
7

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

Código no Github

ning
fonte
6

Eu tenho uma pequena correção para o método de execução de classe CronTab sugerido por Brian .

O tempo terminou em um segundo, levando a um ciclo difícil de um segundo 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()
benc
fonte
4

Não existe uma maneira "python pura" de fazer isso, porque algum outro processo teria que iniciar o python para executar sua solução. Cada plataforma terá uma ou vinte maneiras diferentes de iniciar processos e monitorar seu progresso. Em plataformas unix, o cron é o padrão antigo. No Mac OS X, também existe o launchd, que combina o lançamento do tipo cron com a funcionalidade watchdog que pode manter seu processo ativo, se é isso que você deseja. Depois que o python estiver em execução, você poderá usar o módulo de agendamento para agendar tarefas.

usuario
fonte
4

Sei que há muitas respostas, mas outra solução poderia ser a de decoradores . Este é um exemplo para repetir uma função todos os dias em um horário específico. O melhor de usar dessa maneira é que você só precisa adicionar o Sugar Syntactic Sugar à função que deseja agendar:

@repeatEveryDay(hour=6, minutes=30)
def sayHello(name):
    print(f"Hello {name}")

sayHello("Bob") # Now this function will be invoked every day at 6.30 a.m

E o decorador terá a seguinte aparência:

def repeatEveryDay(hour, minutes=0, seconds=0):
    """
    Decorator that will run the decorated function everyday at that hour, minutes and seconds.
    :param hour: 0-24
    :param minutes: 0-60 (Optional)
    :param seconds: 0-60 (Optional)
    """
    def decoratorRepeat(func):

        @functools.wraps(func)
        def wrapperRepeat(*args, **kwargs):

            def getLocalTime():
                return datetime.datetime.fromtimestamp(time.mktime(time.localtime()))

            # Get the datetime of the first function call
            td = datetime.timedelta(seconds=15)
            if wrapperRepeat.nextSent == None:
                now = getLocalTime()
                wrapperRepeat.nextSent = datetime.datetime(now.year, now.month, now.day, hour, minutes, seconds)
                if wrapperRepeat.nextSent < now:
                    wrapperRepeat.nextSent += td

            # Waiting till next day
            while getLocalTime() < wrapperRepeat.nextSent:
                time.sleep(1)

            # Call the function
            func(*args, **kwargs)

            # Get the datetime of the next function call
            wrapperRepeat.nextSent += td
            wrapperRepeat(*args, **kwargs)

        wrapperRepeat.nextSent = None
        return wrapperRepeat

    return decoratorRepeat
Damia Fuentes
fonte
1

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

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

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)
raph.amiard
fonte
1

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

# 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
fdb
fonte
11
Tenha cuidado ao postar copiar e colar respostas padrão / textuais para várias perguntas, pois elas tendem a ser sinalizadas como "spam" pela comunidade. Se você estiver fazendo isso, geralmente significa que as perguntas são duplicadas; portanto, sinalize-as como tal: stackoverflow.com/a/12360556/419
Kev
1

Se você estiver procurando por um agendador distribuído, consulte https://github.com/sherinkurian/mani - ele precisa de redis, embora possa não ser o que você está procurando. (observe que eu sou o autor) isso foi criado para garantir a tolerância a falhas, com o relógio sendo executado em mais de um nó.

shrnkrn
fonte
0

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

A única preocupação para uma solução python é que as suas necessidades de trabalho para estar sempre em execução e, possivelmente, ser automaticamente "ressuscitado" após uma reinicialização, algo para o qual você fazer necessidade de contar com soluções dependentes do sistema.

Davide
fonte
3
Roll your own é uma opção - embora o melhor código seja o código que você não precisa escrever. A ressurreição, suponho, é algo que talvez eu precise considerar.
jamesh
0

Você pode conferir os [1] Crons do PiCloud [2], mas observe que seus trabalhos não serão executados em sua própria máquina. Também é um serviço que você precisará pagar se usar mais de 20 horas de computação por mês.

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

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

BrainCore
fonte
0

Método do Crontab no servidor.

Nome do arquivo Python hello.py

Etapa 1: Crie um arquivo sh, deixe o nome s.sh

python3 /home/ubuntu/Shaurya/Folder/hello.py> /home/ubuntu/Shaurya/Folder/log.txt 2> & 1

Etapa 2: Abrir o Crontab Editor

crontab -e

Etapa 3: adicionar horário da programação

Use a formatação Crontab

2 * * * * sudo sh /home/ubuntu/Shaurya/Folder/s.sh

Este cron será executado "No minuto 2."

shaurya uppal
fonte
0

Eu gosto de como o pacote pycron resolve esse problema.

import pycron
import time

while True:
    if pycron.is_now('0 2 * * 0'):   # True Every Sunday at 02:00
        print('running backup')
    time.sleep(60)
Duffau
fonte
11
Esta não é uma boa ideia, porque o seu código "print ('running backup')" será iniciado por um minuto inteiro com intervalo de 5s. Portanto, neste caso, o atraso deve ser de 60 segundos.
n158 25/07/19
Você está certo! Obrigado por apontar isso.
Duffau 27/07/19