programar una cita repetitiva en Python 3

Question

que estoy tratando de programar una cita repetitiva para ejecutar cada minuto en Python 3.

que he visto la clase sched.scheduler pero me pregunto si hay otra manera de hacerlo. He oído mencionar que podría usar múltiples hilos para esto, lo que no me importaría hacer.

Básicamente, estoy solicitando un poco de JSON y luego analizándolo; su valor cambia con el tiempo

Para utilizar sched.scheduler tengo que crear un bucle para solicitarlo para programar el incluso para correr durante una hora:

scheduler = sched.scheduler(time.time, time.sleep) 

# Schedule the event. THIS IS UGLY! 
for i in range(60): 
    scheduler.enter(3600 * i, 1, query_rate_limit,()) 

scheduler.run() 

¿Qué otras maneras de hacer esto hay?

Answer 1

Puede usar threading.Timer, pero eso también programa un evento único, de manera similar al método .enter de objetos del planificador.

El patrón normal (en cualquier idioma) para transformar un planificador único en un planificador periódico es hacer que cada evento vuelva a programarse en el intervalo especificado. Por ejemplo, con sched, yo no utilizar un bucle que estás haciendo, sino más bien algo así como:

def periodic(scheduler, interval, action, actionargs=()): 
    scheduler.enter(interval, 1, periodic, 
        (scheduler, interval, action, actionargs)) 
    action(*actionargs) 

e iniciar todo el "horario para siempre periódica" con una llamada

periodic(scheduler, 3600, query_rate_limit) 

O , Podría usar threading.Timer en lugar de scheduler.enter, pero el patrón es bastante similar.

Si necesita una variación más refinada (por ejemplo, detener la reprogramación periódica en un momento dado o bajo ciertas condiciones), no es demasiado difícil acomodarse con algunos parámetros adicionales.

Answer 2

Mi humilde opinión sobre el tema:

from threading import Timer 

class RepeatedTimer(object): 
    def __init__(self, interval, function, *args, **kwargs): 
     self._timer  = None 
     self.function = function 
     self.interval = interval 
     self.args  = args 
     self.kwargs  = kwargs 
     self.is_running = False 
     self.start() 

    def _run(self): 
     self.is_running = False 
     self.start() 
     self.function(*self.args, **self.kwargs) 

    def start(self): 
     if not self.is_running: 
      self._timer = Timer(self.interval, self._run) 
      self._timer.start() 
      self.is_running = True 

    def stop(self): 
     self._timer.cancel() 
     self.is_running = False 

de uso:

from time import sleep 

def hello(name): 
    print "Hello %s!" % name 

print "starting..." 
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start() 
try: 
    sleep(5) # your long-running job goes here... 
finally: 
    rt.stop() # better in a try/finally block to make sure the program ends! 

Características:

única
• biblioteca estándar, sin dependencias externas
• utiliza el patrón sugerido por Alex Martnelli
• start() y stop() son seguros para llamar varias veces, incluso si el temporizador ya ha comenzado/detuvo
• función a ser llamada puede tener argumentos posicionales y nombrados
• Puede cambiar interval en cualquier momento, será efectiva después de la próxima ejecución. Lo mismo para args, kwargs e incluso function!

Answer 3

Puede utilizar el Advanced Python Scheduler. Incluso tiene una interfaz tipo cron.

Answer 4

Puede usar schedule. Funciona en Python 2.7 y 3.3 y es bastante ligera:

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) 

Answer 5

Basado en respuesta MestreLion, que resolver un pequeño problema con multihilo:

from threading import Timer, Lock 


class Periodic(object): 
    """ 
    A periodic task running in threading.Timers 
    """ 

    def __init__(self, interval, function, *args, **kwargs): 
     self._lock = Lock() 
     self._timer = None 
     self.function = function 
     self.interval = interval 
     self.args = args 
     self.kwargs = kwargs 
     self._stopped = True 
     if kwargs.pop('autostart', True): 
      self.start() 

    def start(self, from_run=False): 
     self._lock.acquire() 
     if from_run or self._stopped: 
      self._stopped = False 
      self._timer = Timer(self.interval, self._run) 
      self._timer.start() 
      self._lock.release() 

    def _run(self): 
     self.start(from_run=True) 
     self.function(*self.args, **self.kwargs) 

    def stop(self): 
     self._lock.acquire() 
     self._stopped = True 
     self._timer.cancel() 
     self._lock.release() 

Answer 6

Uso Celery.

from celery.task import PeriodicTask 
from datetime import timedelta 


class ProcessClicksTask(PeriodicTask): 
    run_every = timedelta(minutes=30) 

    def run(self, **kwargs): 
     #do something 

Answer 7

tarea programador es un programador en proceso de organizar y ejecutar la tarea periódicamente de acuerdo con el archivo de configuración YAML. Por favor remítase a https://github.com/tzutalin/task-scheduler

Answer 8

Sobre la base de la respuesta de Alex Martelli, he implementado versión decorador que es más fácil integrado.

import sched 
import time 
import datetime 
from functools import wraps 
from threading import Thread 


def async(func): 
    @wraps(func) 
    def async_func(*args, **kwargs): 
     func_hl = Thread(target=func, args=args, kwargs=kwargs) 
     func_hl.start() 
     return func_hl 
    return async_func 


def schedule(interval): 
    def decorator(func): 
     def periodic(scheduler, interval, action, actionargs=()): 
      scheduler.enter(interval, 1, periodic, 
          (scheduler, interval, action, actionargs)) 
      action(*actionargs) 

     @wraps(func) 
     def wrap(*args, **kwargs): 
      scheduler = sched.scheduler(time.time, time.sleep) 
      periodic(scheduler, interval, func) 
      scheduler.run() 
     return wrap 
    return decorator 


@async 
@schedule(1) 
def periodic_event(): 
    print(datetime.datetime.now()) 


if __name__ == '__main__': 
    print('start') 
    periodic_event() 
    print('end')