from starlette.middleware.base import BaseHTTPMiddleware

from fastapi import Request

import json

from fastapi.responses import JSONResponse,StreamingResponse

import traceback

class ResponseMiddleware(BaseHTTPMiddleware):

    async def dispatch(self,request:Request,call_next):

        try:

            response=await call_next(request)

            if response.status_code==200:

                 # 处理流式响应

                 body = b"".join([chunk async for chunk in response.body_iterator])

                 response.body_iterator = iter([body])  # 重置迭代器

                 try:

                    body_data = json.loads(body.decode()) if body else None

                 except json.JSONDecodeError:

                    body_data = body.decode()  # 非JSON则保留原始数据

                 return JSONResponse({

                    "code":200,

                    "data":body_data,

                    "message":"success"

                })

            return response

        except Exception as e:

            return JSONResponse({

                    "code":500,

                    "data":None,

                    "message":traceback.format_exception(e) @跟踪错误信息

            },status_code=500)

 

使用 方法

#main.py

from app.utils.responseMiddlerWare import ResponseMiddleware

app=FASTAPI()

app.addmiddleware(ResponseMiddleware)

如果想过滤掉其中的某个路径信息，可以这样写

import os

def fitler_trackback(exc:BaseException):

    tb=exc.__traceback__

    while tb:

        frame=tb.tb_frame

        # 检查文件名或模块路径是否含中间件特征

        frame_file = os.path.abspath(frame.f_code.co_filename).lower()

        is_middleware = (

            'middleware' in frame_file or

             '/middlewares/' in frame_file or

             'starlette' in frame_file or

            frame.f_code.co_name == 'dispatch'  # 常见中间件入口方法

        )

        if not is_middleware:

           break

        tb=tb.tb_next

    return traceback.format_exception(type(exc),exc,tb)

 引用

from starlette.middleware.base import BaseHTTPMiddleware

from fastapi import Request

import json

from fastapi.responses import JSONResponse,StreamingResponse

import traceback

class ResponseMiddleware(BaseHTTPMiddleware):

    async def dispatch(self,request:Request,call_next):

        try:

            response=await call_next(request)

            if response.status_code==200:

                 # 处理流式响应

                 body = b"".join([chunk async for chunk in response.body_iterator])

                 response.body_iterator = iter([body])  # 重置迭代器

                 try:

                    body_data = json.loads(body.decode()) if body else None

                 except json.JSONDecodeError:

                    body_data = body.decode()  # 非JSON则保留原始数据

                 return JSONResponse({

                    "code":200,

                    "data":body_data,

                    "message":"success"

                })

            return response

        except Exception as e:

            return JSONResponse({

                    "code":500,

                    "data":None,

                    "message":fitler_trackback(e) @跟踪错误信息

            },status_code=500)