Comment enregistrer efficacement les corps de requête/réponse HTTP bruts dans FastAPI ?

Journalisation des requêtes/réponses HTTP brutes dans Python FastAPI

Exigence :

Capturez et enregistrez les corps JSON bruts de fichiers spécifiques acheminer les requêtes et les réponses, avec une taille de données d'environ 1 Mo, sans affecter de manière significative les temps de réponse.

Option 1 : Utilisation du middleware

Concept du middleware

Le middleware intercepte chaque requête avant qu'elle n'atteigne les points de terminaison et les réponses avant qu'elles ne soient transmises aux clients, permettant ainsi la manipulation des données. Cependant, le problème lié à la consommation du flux de corps de la requête dans le middleware est qu'il devient indisponible pour les points de terminaison en aval. Par conséquent, nous utiliserons la fonction set_body() pour la rendre disponible.

Pour les réponses, utiliser BackgroundTask

La journalisation peut être effectuée à l'aide de BackgroundTask, qui garantit que la journalisation a lieu après que la réponse a été envoyé au client, évitant ainsi les retards dans les temps de réponse.

Exemple de middleware

# Logging middleware
async def some_middleware(request: Request, call_next):
    req_body = await request.body()
    await set_body(request, req_body)
    response = await call_next(request)

    # Body storage in RAM
    res_body = b''
    async for chunk in response.body_iterator:
        res_body += chunk

    # Background logging task
    task = BackgroundTask(log_info, req_body, res_body)
    return Response(...)

# Endpoint using middleware
@app.post('/')
async def main(payload: Dict):
    pass

Option 2 : Personnalisé Classe APIRoute

Extension de classe APIRoute

En créant une classe APIRoute personnalisée, nous pouvons contrôler les corps de requête et de réponse, limitant son utilisation à des itinéraires spécifiques via un APIRoute.

Considérations importantes

Pour les réponses volumineuses (par exemple, streaming multimédia), l'itinéraire personnalisé peut rencontrer des problèmes de RAM ou des retards côté client en raison de la lecture. toute la réponse dans la RAM. Par conséquent, envisagez d'exclure ces points de terminaison de la route personnalisée.

Exemple de classe APIRoute personnalisée

class LoggingRoute(APIRoute):
    async def custom_route_handler(request: Request) -> Response:
        req_body = await request.body()
        response = await original_route_handler(request)

        # Response handling based on type
        if isinstance(response, StreamingResponse):
            res_body = b''
            async for item in response.body_iterator:
                res_body += item
            response = Response(...)
        else:
            response.body

        # Logging task
        task = BackgroundTask(log_info, req_body, response.body)
        response.background = task
        return response

# Endpoint using custom APIRoute
@router.post('/')
async def main(payload: Dict):
    return payload

Choisir une option

Les deux options fournissent des solutions pour enregistrer les données de demande et de réponse sans impactant considérablement les délais de réponse. L'option 1 permet une journalisation générale, tandis que l'option 2 offre un contrôle granulaire sur les itinéraires qui nécessitent une journalisation.

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!