How to Efficiently Run Background Jobs in FastAPI?

Background Jobs in FastAPI: Running Functions Asynchronously

In FastAPI, background functions are necessary when you want to perform asynchronous operations that should not block or delay the main API operations. Let's explore how to set up background jobs effectively.

1. Thread-Based Background Tasks:

Starting threads is a straightforward way to run functions in the background:

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

However, this approach has a limitation: it blocks the main thread of execution. To avoid this, start your thread before the main API process as seen below:

_worker_thread.start()
uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

2. Event-Based Background Tasks:

Another option is to use event-based schedulers:

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)

3. AsyncIO Background Tasks:

If your background function is async, you can use the asyncio.create_task() function:

from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio


async def print_task(s): 
    while True:
        print('Hello')
        await asyncio.sleep(s)

        
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Run at startup
    asyncio.create_task(print_task(5))
    yield
    # Run on shutdown (if required)
    print('Shutting down...')


app = FastAPI(lifespan=lifespan)

The above is the detailed content of How to Efficiently Run Background Jobs in FastAPI?. For more information, please follow other related articles on the PHP Chinese website!