¿Cómo ejecutar repetidamente una función cada x segundos?

Question 1

Quiero ejecutar repetidamente una función en Python cada 60 segundos para siempre (al igual que un NSTemporizador en Objective C o setTimeout en JS). Este código se ejecutará como un demonio y es efectivamente como llamar a la secuencia de comandos de python cada minuto usando un cron, pero sin necesidad de que el usuario lo configure.

En esta pregunta sobre un cron implementado en Python, la solución parece efectivamente solo dormir() durante x segundos. No necesito una funcionalidad tan avanzada, así que tal vez algo como esto funcione

while True:
    # Code executed here
    time.sleep(60)

¿Hay algún problema previsible con este código?

Question 2

Si su programa aún no tiene un bucle de eventos, use el programado módulo, que implementa un programador de eventos de propósito general.

import sched, time
s = sched.scheduler(time.time, time.sleep)
def do_something(sc): 
    print("Doing stuff...")
    # do your stuff
    sc.enter(60, 1, do_something, (sc,))

s.enter(60, 1, do_something, (s,))
s.run()

Si ya está utilizando una biblioteca de bucles de eventos como asyncio, trio, tkinter, PyQt5, gobject, kivyy muchos otros: simplemente programe la tarea utilizando los métodos de su biblioteca de bucle de eventos existente.

Question 3

Bloquee su ciclo de tiempo al reloj del sistema de esta manera:

import time
starttime = time.time()
while True:
    print("tick")
    time.sleep(60.0 - ((time.time() - starttime) % 60.0))

Question 4

Si desea una forma sin bloqueo de ejecutar su función periódicamente, en lugar de un bucle infinito de bloqueo, usaría un temporizador de subprocesos. De esta manera, su código puede seguir ejecutándose y realizar otras tareas y aún así llamar a su función cada n segundos. Utilizo mucho esta técnica para imprimir información de progreso en tareas largas e intensivas de CPU/disco/red.

Aquí está el código que publiqué en una pregunta similar, con el control start() y stop():

from threading import Timer

class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.interval   = interval
        self.function   = function
        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!

Características:

Solo biblioteca estándar, sin dependencias externas
start() y stop() es seguro llamar varias veces incluso si el temporizador ya se ha iniciado/detenido
la función a llamar puede tener argumentos posicionales y con nombre
Tu puedes cambiar interval en cualquier momento, será efectivo después de la próxima ejecución. Igual por args, kwargs e incluso function!

Question 5

Es posible que desee considerar Retorcido que es una biblioteca de red de Python que implementa el Patrón de reactores.

from twisted.internet import task, reactor

timeout = 60.0 # Sixty seconds

def doWork():
    #do work here
    pass

l = task.LoopingCall(doWork)
l.start(timeout) # call every sixty seconds

reactor.run()

Si bien “while True: sleep(60)” probablemente funcionará, es probable que Twisted ya implemente muchas de las características que eventualmente necesitará (daemonización, registro o manejo de excepciones como lo señala bobince) y probablemente sea una solución más robusta.

Question 6

Aquí hay una actualización del código de MestreLion que evita la deriva con el tiempo.

La clase RepeatedTimer aquí llama a la función dada cada “intervalo” segundos según lo solicitado por el OP; la programación no depende de cuánto tiempo se tarde en ejecutar la función. Me gusta esta solución porque no tiene dependencias de bibliotecas externas; esto es solo Python puro.

import threading 
import time

class RepeatedTimer(object):
  def __init__(self, interval, function, *args, **kwargs):
    self._timer = None
    self.interval = interval
    self.function = function
    self.args = args
    self.kwargs = kwargs
    self.is_running = False
    self.next_call = time.time()
    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.next_call += self.interval
      self._timer = threading.Timer(self.next_call - time.time(), self._run)
      self._timer.start()
      self.is_running = True

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

Ejemplo de uso (copiado de la respuesta de MestreLion):

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!

Question 7

import time, traceback

def every(delay, task):
  next_time = time.time() + delay
  while True:
    time.sleep(max(0, next_time - time.time()))
    try:
      task()
    except Exception:
      traceback.print_exc()
      # in production code you might want to have this instead of course:
      # logger.exception("Problem while executing repetitive task.")
    # skip tasks if we are behind schedule:
    next_time += (time.time() - next_time) // delay * delay + delay

def foo():
  print("foo", time.time())

every(5, foo)

Si desea hacer esto sin bloquear el código restante, puede usarlo para permitir que se ejecute en su propio hilo:

import threading
threading.Thread(target=lambda: every(5, foo)).start()

Esta solución combina varias características que rara vez se encuentran combinadas en las otras soluciones:

Manejo de excepciones: En la medida de lo posible, en este nivel, las excepciones se manejan correctamente, es decir, se registran con fines de depuración sin abortar nuestro programa.
Sin encadenamiento: La implementación similar a una cadena común (para programar el próximo evento) que encuentra en muchas respuestas es frágil en el aspecto de que si algo sale mal dentro del mecanismo de programación (threading.Timer o lo que sea), esto terminará la cadena. No habrá más ejecuciones entonces, incluso si la razón del problema ya está solucionada. Un bucle simple y esperando con un simple sleep() es mucho más robusto en comparación.
Sin deriva: Mi solución mantiene un registro exacto de los tiempos en los que se supone que debe ejecutarse. No hay deriva en función del tiempo de ejecución (como en muchas otras soluciones).
Salto a la comba: Mi solución omitirá tareas si una ejecución tomó demasiado tiempo (por ejemplo, haga X cada cinco segundos, pero X tomó 6 segundos). Este es el comportamiento estándar de cron (y por una buena razón). Muchas otras soluciones simplemente ejecutan la tarea varias veces seguidas sin demora. En la mayoría de los casos (p. ej., tareas de limpieza), esto no se desea. Si se es deseado, simplemente use next_time += delay en cambio.

Question 8

La forma más fácil que creo que es:

import time

def executeSomething():
    #code here
    time.sleep(60)

while True:
    executeSomething()

De esta manera se ejecuta su código, luego espera 60 segundos y luego se ejecuta de nuevo, espera, ejecuta, etc. No hay necesidad de complicar las cosas 😀

¿Ha sido útil esta solución?