tutorial - Programe un evento repetitivo en Python 3

while en python 3

Question

Estoy intentando programar un evento repetitivo para ejecutar cada minuto en Python 3.

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

Básicamente solicito un poco de JSON y luego lo analizo; su valor cambia con el tiempo

Para usar sched.scheduler tengo que crear un bucle para solicitar que programe la ejecución par para 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 formas de hacer esto hay?

Answer 1

Aquí hay un loop rápido y sucio sin bloqueo con Thread :

#!/usr/bin/env python3
import threading,time
def worker():
print(time.time())
time.sleep(5)
t = threading.Thread(target=worker)
t.start()
threads = []
t = threading.Thread(target=worker)
threads.append(t)
t.start()
time.sleep(7)
print("Hello World")

No hay nada particularmente especial, el worker crea un nuevo hilo de sí mismo con un retraso. Puede que no sea lo más eficiente, pero lo suficientemente simple. La respuesta de Northtree sería el camino a seguir si necesita una solución más sofisticada.

Y en base a this , podemos hacer lo mismo, solo con Timer :

#!/usr/bin/env python3
import threading,time
def hello():
t = threading.Timer(10.0, hello)
t.start()
print( "hello, world",time.time() )
t = threading.Timer(10.0, hello)
t.start()
time.sleep(12)
print("Oh,hai",time.time())
time.sleep(4)
print("How''s it going?",time.time())

Answer 2

Basado en la respuesta de Alex Martelli, he implementado una versión de decorador que es más fácil de integrar.

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

Answer 3

Basado en la respuesta de MestreLion, resuelve 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 4

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

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!

caracteristicas:

• Biblioteca estándar solamente, 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 / detenido
• la función a ser llamada puede tener argumentos posicionales y nombrados
• Puede cambiar el interval cualquier momento, será efectivo después de la próxima ejecución. ¡Lo mismo para args , kwargs e incluso function !

Answer 5

Podría usar threading.Timer , pero eso también programa un evento único, de manera similar al método .enter de los 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 se vuelva a programar en el intervalo especificado. Por ejemplo, con sched , no usaría un bucle como lo está haciendo, sino algo así como:

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

e inicie el "programa siempre para siempre" con una llamada

periodic(scheduler, 3600, query_rate_limit)

O bien, podría usar threading.Timer vez 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 6

Podría usar el Programador avanzado de Python . Incluso tiene una interfaz tipo cron.

Answer 7

Puedes usar el schedule Funciona en Python 2.7 y 3.3 y es bastante liviano:

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 8

Usa 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 9

Ver mi muestra

import sched, time
def myTask(m,n):
print n+'' ''+m
def periodic_queue(interval,func,args=(),priority=1):
s = sched.scheduler(time.time, time.sleep)
periodic_task(s,interval,func,args,priority)
s.run()
def periodic_task(scheduler,interval,func,args,priority):
func(*args)
scheduler.enter(interval,priority,periodic_task,
(scheduler,interval,func,args,priority))
periodic_queue(1,myTask,(''world'',''hello''))

Answer 10

task-scheduler es un programador en proceso para organizar y ejecutar la tarea periódicamente según el archivo de configuración YAML. Por favor, consulte https://github.com/tzutalin/task-scheduler