FastAPI python: How to run a thread in the background?-CodePudding

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

CodePudding user response：

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)