How to Run Background Tasks in FastAPI?-Python Tutorial-php.cn

How to Run Background Tasks in FastAPI?

Susan Sarandon

Release： 2024-12-18 12:03:18

Original

554 people have browsed it

How to Run Background Tasks in FastAPI?

Background Task Execution in FastAPI

In FastAPI, developers may encounter a scenario where they need to run a function in the background, unrelated to their API. To achieve this, it's necessary to employ threading mechanisms.

Solution 1: Starting Thread before uvicorn.run

One approach involves starting the thread before invoking uvicorn.run, as uvicorn.run blocks the thread. This approach can be implemented as follows:

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=&quot;0.0.0.0&quot;, port=8000)

Copy after login

Alternatively, developers can leverage FastAPI's startup event to initiate the thread before the application's commencement:

@app.on_event(&quot;startup&quot;)
async def startup_event():
    t = BackgroundTasks()
    t.start()

Copy after login

Solution 2: Event Scheduler

Another option is to implement a recurring Event scheduler for the background task:

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(&quot;Hello&quot;)
    sc.enter(5, 1, print_event, (sc,))

def start_scheduler():
    s.enter(5, 1, print_event, (s,))
    s.run()

@app.on_event(&quot;startup&quot;)
async def startup_event():
    thread = Thread(target=start_scheduler)
    thread.start()

if __name__ == '__main__':
    uvicorn.run(app, host=&quot;0.0.0.0&quot;, port=8000)

Copy after login

Solution 3: Using the Event Loop

For asynchronous background tasks (async def functions), developers can utilize the current event loop with the asyncio.create_task() function:

from asyncio import create_task
from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_task():
    await create_task(print_task(5))

Copy after login

In this approach, the event loop is created when initiating the uvicorn server.

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