Comment initialiser et réutiliser efficacement un objet global sur les points de terminaison FastAPI ?

Comment initialiser un objet ou une variable globale et le réutiliser dans chaque point de terminaison FastAPI ?

Contexte

Considérez une classe personnalisée conçue pour l'envoi de notifications, dont l'initialisation implique l'établissement d'une connexion à un serveur de notification, un processus qui prend beaucoup de temps. Cette classe est utilisée dans une tâche en arrière-plan dans FastAPI pour éviter de retarder la réponse du point de terminaison. Cependant, l'approche actuelle présente des limites :

file1.py:
noticlient = NotificationClient()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, noticlient)
    return result

file2.py:
def add_some_tasks(data, background_tasks: BackgroundTasks, noticlient):
    background_tasks.add_task(noticlient.send, param1, param2)
    result = some_operation
    return result

L'initialisation globale de NotificationClient dans file1.py entraîne plusieurs initialisations redondantes à chaque fois qu'une requête est reçue, ce qui est inefficace.

Approches

Option 1 : Utiliser app.state

FastAPI permet vous devez stocker un état arbitraire à l'aide de app.state. Vous pouvez initialiser l'objet NotificationClient et l'ajouter à app.state lors du démarrage ou de l'arrêt de FastAPI à l'aide d'une fonction de cycle de vie de dépendance telle que lifespan.

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise the Client and add it to app.state
    '''
    app.state.n_client = NotificationClient()
    yield
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    app.state.n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.app.state.n_client
    # ...

Option 2 : Utiliser la durée de vie de Starlette

Le gestionnaire de durée de vie de Starlette vous permet de définir des objets d'état accessibles au sein des points de terminaison via demande.état.

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise the Client and add it to request.state
    '''
    n_client = NotificationClient()
    yield {'n_client': n_client}
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.state.n_client
    # ...

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!