Estoy creando un servidor en python usando FastAPI, y quiero que una función que no esté relacionada con mi API se ejecute en segundo plano cada 5 minutos (como verificar cosas de una API e imprimir cosas dependiendo de la respuesta)
Intenté crear un hilo que ejecute la función de inicio de trabajo, pero no imprime nada.
¿Alguien sabe cómo hacerlo?
def start_worker(): print('[main]: starting worker...') my_worker = worker.Worker() my_worker.working_loop() # this function prints "hello" every 5 seconds if __name__ == '__main__': print('[main]: starting...') uvicorn.run(app, host="0.0.0.0", port=8000, reload=True) _worker_thread = Thread(target=start_worker, daemon=False) _worker_thread.start()Debe iniciar su subproceso antes de llamar a uvicorn.run , ya que uvicorn.run está bloqueando el subproceso.
PD: En su pregunta, indica que desea que la tarea en segundo plano se ejecute cada 5 minutos, pero en su código dice cada 5 segundos . Los siguientes ejemplos asumen que es lo último que desea. Si desea que se ejecute cada 5 minutos, ajuste el tiempo a 60 * 5 .
Opción 1
import time import threading from fastapi import FastAPI import uvicorn app = FastAPI() class BackgroundTasks(threading.Thread): def run(self,*args,**kwargs): while True: print('Hello') time.sleep(5) if __name__ == '__main__': t = BackgroundTasks() t.start() uvicorn.run(app, host="0.0.0.0", port=8000)También puede iniciar su subproceso utilizando el evento de inicio de FastAPI, siempre que esté bien ejecutarlo antes de que se inicie la aplicación.
@app.on_event("startup") async def startup_event(): t = BackgroundTasks() t.start()opcion 2
En su lugar, podría usar un programador de eventos repetitivo para la tarea en segundo plano, como se muestra a continuación:
import sched, time from threading import Thread from fastapi import FastAPI import uvicorn app = FastAPI() s = sched.scheduler(time.time, time.sleep) def print_event(sc): print("Hello") sc.enter(5, 1, print_event, (sc,)) def start_scheduler(): s.enter(5, 1, print_event, (s,)) s.run() @app.on_event("startup") async def startup_event(): thread = Thread(target = start_scheduler) thread.start() if __name__ == '__main__': uvicorn.run(app, host="0.0.0.0", port=8000)