Wie führe ich Hintergrundaufgaben in FastAPI aus?

Ausführung von Hintergrundaufgaben in FastAPI

In FastAPI können Entwickler auf ein Szenario stoßen, in dem sie eine Funktion im Hintergrund ausführen müssen, die nichts damit zu tun hat ihre API. Um dies zu erreichen, müssen Threading-Mechanismen eingesetzt werden.

Lösung 1: Thread vor uvicorn.run starten

Ein Ansatz besteht darin, den Thread vor dem Aufruf von uvicorn.run zu starten. da uvicorn.run den Thread blockiert. Dieser Ansatz kann wie folgt implementiert werden:

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)

Alternativ können Entwickler das Startereignis von FastAPI nutzen, um den Thread vor dem Start der Anwendung zu initiieren:

@app.on_event("startup")
async def startup_event():
    t = BackgroundTasks()
    t.start()

Lösung 2: Ereignis Planer

Eine weitere Möglichkeit besteht darin, einen wiederkehrenden Ereignisplaner für den Hintergrund zu implementieren Aufgabe:

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)

Lösung 3: Verwendung der Ereignisschleife

Für asynchrone Hintergrundaufgaben (asynchrone Def-Funktionen) können Entwickler die aktuelle Ereignisschleife mit dem Asyncio nutzen .create_task()-Funktion:

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

Bei diesem Ansatz wird die Ereignisschleife beim Initiieren des Uvicorns erstellt Server.

Das obige ist der detaillierte Inhalt vonWie führe ich Hintergrundaufgaben in FastAPI aus?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!