Wie kann man rohe HTTP-Anforderungs- und Antwortkörper in FastAPI effizient protokollieren?

So zeichnen Sie rohe HTTP-Anfragen und -Antworten in Python FastAPI auf

Einführung:

In Um die Prüfanforderungen für Ihren Python FastAPI-basierten Webdienst zu erfüllen, müssen Sie die rohen JSON-Texte sowohl von Anfragen als auch von Antworten auf bestimmten Routen beibehalten. In diesem Leitfaden werden zwei praktikable Lösungen vorgestellt, um dies zu erreichen, ohne die Antwortzeiten spürbar zu beeinträchtigen, selbst wenn mit Körpergrößen von etwa 1 MB gearbeitet wird.

Option 1: Middleware-Nutzung

Middleware-Mechanik:

Middleware fungiert als Gatekeeper für Anfragen, die in die Anwendung gelangen. Es ermöglicht die Bearbeitung von Anfragen vor der Endpunktverarbeitung und Antworten vor der Rückgabe an Clients. Sie können Middleware mithilfe des @app.middleware-Dekorators für eine Funktion einrichten:

Verwaltung des Anforderungs- und Antworttexts:

Zugriff auf den Anforderungstext aus dem Stream innerhalb der Middleware ( Wenn Sie request.body() oder request.stream() verwenden, müssen Sie es später im Anfrage-Antwort-Zyklus verfügbar machen. Im verlinkten Beitrag wird diese Problemumgehung besprochen, die jetzt für FastAPI-Versionen 0.108.0 und höher nicht mehr erforderlich ist.

Für den Antworttext können Sie die in diesem Beitrag beschriebene Technik replizieren, um den Text direkt zu konsumieren und zurückzugeben und den Status anzugeben Code, Header und Medientyp zusammen mit der ursprünglichen Antwort.

Protokollierung Daten:

Verwenden Sie BackgroundTask, um Daten zu protokollieren und sicherzustellen, dass sie nach Abschluss der Antwort ausgeführt werden. Dadurch entfällt das Warten des Clients auf Protokollierungsaufgaben und die Integrität der Antwortzeit bleibt erhalten.

Option 2: Benutzerdefinierte APIRoute-Implementierung

Benutzerdefinierte APIRoute:

Diese Option beinhaltet das Erstellen einer benutzerdefinierten APIRoute-Klasse zum Bearbeiten von Anforderungs- und Antworttexten vor der Verarbeitung von Endpunkten oder der Rückgabe von Ergebnissen an Clients. Es ermöglicht die Isolierung der benutzerdefinierten Routenbehandlung zu bestimmten Endpunkten durch die Verwendung eines dedizierten APIRouter:

Überlegungen:

Speicherbeschränkungen:

Bei beiden Ansätzen kann es zu Problemen kommen, wenn große Anfrage- oder Antworttexte den verfügbaren Server-RAM überschreiten. Das Streamen großer Antworten kann zu clientseitigen Verzögerungen oder Reverse-Proxy-Fehlern führen. Beschränken Sie die Middleware-Nutzung auf bestimmte Routen oder schließen Sie Endpunkte mit großen Streaming-Antworten aus, um potenzielle Probleme zu vermeiden.

Beispielcode (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

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

Diese Lösungen bieten effiziente Methoden zum Protokollieren von Roh-HTTP-Anfragen und Antwortkörper in FastAPI, ohne die Antwortzeiten wesentlich zu beeinträchtigen.

Das obige ist der detaillierte Inhalt vonWie kann man rohe HTTP-Anforderungs- und Antwortkörper in FastAPI effizient protokollieren?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!