Comment exécuter des tâches en arrière-plan dans FastAPI ?

Exécution de tâches en arrière-plan dans FastAPI

Dans FastAPI, les développeurs peuvent rencontrer un scénario dans lequel ils doivent exécuter une fonction en arrière-plan, sans rapport avec leur API. Pour y parvenir, il est nécessaire d'utiliser des mécanismes de threading.

Solution 1 : Démarrer le fil avant uvicorn.run

Une approche consiste à démarrer le fil de discussion avant d'invoquer uvicorn.run, car uvicorn.run bloque le fil. Cette approche peut être mise en œuvre comme suit :

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)

Alternativement, les développeurs peuvent tirer parti de l'événement de démarrage de FastAPI pour lancer le fil de discussion avant le démarrage de l'application :

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

Solution 2 : Événement Planificateur

Une autre option consiste à implémenter un planificateur d'événements récurrents pour l'arrière-plan tâche :

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)

Solution 3 : Utilisation de la boucle d'événements

Pour les tâches en arrière-plan asynchrones (fonctions de définition asynchrones), les développeurs peuvent utiliser la boucle d'événements actuelle avec l'asyncio Fonction .create_task() :

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

Dans cette approche, la boucle d'événements est créée lors du lancement de l'uvicorn serveur.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!