I'm making a server in python using FastAPI, and I want a function that is not related to my API, to run in background every 5 minutes (like checking stuff from an api and print stuff depending on the response)
I've tried to make a thread that run the function start worker but it doesn't print anything.
Does anyone know how to do so ?
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()
You should start your Thread before calling uvicorn.run, as uvicorn.run is blocking the thread.
PS: In your question you state that you would like the background task to run every 5 minutes, but in your code you say every 5 seconds. The below examples assume that is the latter you want. If you want it to be executed every 5 minutes instead, then adjust the time to 60 * 5.
Option 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)
You could also start your thread using FastAPI's startup event, as long as it is ok to run before the application starts.
@app.on_event("startup")
async def startup_event():
t = BackgroundTasks()
t.start()
Option 2
You could instead use a repeating Event scheduler for the background task, as below:
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)