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

Comment enregistrer des requêtes et des réponses HTTP brutes dans Python FastAPI

Introduction :

Dans Afin de répondre aux exigences d'audit de votre service Web basé sur Python FastAPI, vous devez conserver les corps JSON bruts des requêtes et des réponses sur certaines routes. Ce guide présentera deux solutions viables pour y parvenir sans impact notable sur les temps de réponse, même lorsque vous travaillez avec des tailles de corps d'environ 1 Mo.

Option 1 : Utilisation du middleware

Mécanique du middleware :

Le middleware fonctionne comme un contrôleur d'accès pour les requêtes entrant dans l'application. Il permet de traiter les demandes avant le traitement du point final et les réponses avant de retourner aux clients. Vous pouvez établir un middleware à l'aide du décorateur @app.middleware sur une fonction :

Gestion du corps de requête et de réponse :

Pour accéder au corps de la requête à partir du flux dans le middleware ( en utilisant request.body() ou request.stream()), vous devrez le rendre disponible plus tard dans le cycle demande-réponse. L'article lié traite de cette solution de contournement, qui n'est désormais plus nécessaire pour les versions FastAPI 0.108.0 et supérieures.

Pour le corps de la réponse, vous pouvez reproduire la technique décrite dans cet article pour consommer et renvoyer le corps directement, en fournissant le statut code, en-têtes et type de média ainsi que la réponse originale.

Données de journalisation :

Employez BackgroundTask pour enregistrer les données, garantissant leur exécution une fois la réponse terminée. Cela élimine l'attente du client pour les tâches de journalisation et maintient l'intégrité du temps de réponse.

Option 2 : implémentation d'APIRoute personnalisée

APIRoute personnalisée :

Cette option implique la création d'une classe APIRoute personnalisée pour manipuler les corps de requête et de réponse avant de traiter les points de terminaison ou de renvoyer les résultats aux clients. Il permet d'isoler la gestion des routes personnalisées vers des points de terminaison spécifiques à l'aide d'un APIRouter dédié :

Considérations :

Contraintes de mémoire :

Les deux approches peuvent rencontrer des défis avec des corps de requêtes ou de réponses volumineux dépassant la RAM disponible du serveur. La diffusion de réponses volumineuses peut introduire des retards côté client ou des erreurs de proxy inverse. Limitez l'utilisation du middleware à des itinéraires spécifiques ou excluez les points de terminaison avec des réponses de streaming volumineuses pour éviter les problèmes potentiels.

Exemple de code (option 1) :

from fastapi import FastAPI, APIRouter, Response, Request
from starlette.background import BackgroundTask
from fastapi.routing import APIRoute
from starlette.types import Message
from typing import Dict, Any
import logging


app = FastAPI()
logging.basicConfig(filename='info.log', level=logging.DEBUG)


def log_info(req_body, res_body):
    logging.info(req_body)
    logging.info(res_body)



# Not required for FastAPI >= 0.108.0
async def set_body(request: Request, body: bytes):
    async def receive() -> Message:
        return {'type': 'http.request', 'body': body}
    request._receive = receive


@app.middleware('http')
async def some_middleware(request: Request, call_next):
    req_body = await request.body()
    await set_body(request, req_body)  # Not required for FastAPI >= 0.108.0
    response = await call_next(request)
    
    res_body = b''
    async for chunk in response.body_iterator:
        res_body += chunk
    
    task = BackgroundTask(log_info, req_body, res_body)
    return Response(content=res_body, status_code=response.status_code, 
        headers=dict(response.headers), media_type=response.media_type, background=task)


@app.post('/')
def main(payload: Dict[Any, Any]):
    return payload

Exemple Code (Option 2) :

from fastapi import FastAPI, APIRouter, Response, Request
from starlette.background import BackgroundTask
from starlette.responses import StreamingResponse
from fastapi.routing import APIRoute
from starlette.types import Message
from typing import Callable, Dict, Any
import logging
import httpx


def log_info(req_body, res_body):
    logging.info(req_body)
    logging.info(res_body)

       
class LoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            req_body = await request.body()
            response = await original_route_handler(request)
            tasks = response.background
            
            if isinstance(response, StreamingResponse):
                res_body = b''
                async for item in response.body_iterator:
                    res_body += item
                  
                task = BackgroundTask(log_info, req_body, res_body)
                response = Response(content=res_body, status_code=response.status_code, 
                        headers=dict(response.headers), media_type=response.media_type)
            else:
                task = BackgroundTask(log_info, req_body, response.body)
            
            # Check if the original response had background tasks already attached to it
            if tasks:
                tasks.add_task(task)  # Add the new task to the tasks list
                response.background = tasks
            else:
                response.background = task
                
            return response
            
        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=LoggingRoute)
logging.basicConfig(filename='info.log', level=logging.DEBUG)


@router.post('/')
def main(payload: Dict[Any, Any]):
    return payload


@router.get('/video')
def get_video():
    url = 'https://storage.googleapis.com/gtv-videos-bucket/sample/ForBiggerBlazes.mp4'
    
    def gen():
        with httpx.stream('GET', url) as r:
            for chunk in r.iter_raw():
                yield chunk

    return StreamingResponse(gen(), media_type='video/mp4')


app.include_router(router)

Ces solutions fournissent des méthodes efficaces pour enregistrer les requêtes et réponses HTTP brutes corps dans FastAPI sans affecter de manière significative les temps de réponse.

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!