FastAPI自定义日志装饰器引发422 Unprocessable Entity错误

错误详情 #

调用接口时返回422 Unprocessable Entity，响应内容如下：

{
    "detail": [
        {
            "loc": [
                "query",
                "self"
            ],
            "msg": "field required",
            "type": "value_error.missing"
        }
    ]
}

自定义装饰器代码 #

class LogEndpointAction(DatabaseClass):

    def __init__(self, entity_name: str, action: str):
        self.entity_name = entity_name
        self.action = action

    def __call__(self, func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            request: Request = kwargs.get('request')
            print(' args ', args)
            print(' req ', request)
            current_user = await self._authenticate_user(request)
            response = await func(*args, **kwargs)
            body = await request.json() if request.method in ["POST", "PUT", "PATCH"] else {}

            log_data = {
                'tenant_id': '',
                'user_id': '',
                'full_name': '',
                'roles': [],
                'email': '',
                'entity_name': self.entity_name,
                'data': json.dumps(body),
                'i18n_key': self.action
            }

            await self.db.execute(query=INSERT_LOG_QUERY, values=log_data)
            return response


        return wrapper

端点使用方式 #

@LogEndpointAction(entity_name='Sales', action='Get all')
@app.get("/")
async def root():
    return {"message": "Hello World"}

问题原因 #

1. 装饰器顺序错误：FastAPI的路由装饰器（如@app.get）必须是距离端点函数最近的装饰器。你把自定义装饰器放在路由装饰器上方，导致FastAPI错误地将装饰器类的self识别为接口的查询参数，从而抛出字段缺失的错误。
2. Request对象获取不可靠：当前从kwargs.get('request')获取请求对象，但如果端点函数没有显式注入Request参数，kwargs中不会包含该对象，会导致后续_authenticate_user和request.json()调用失败。

修复方案 #

1. 调整装饰器顺序 #

将自定义装饰器放在路由装饰器下方（靠近端点函数），确保FastAPI先识别路由配置，再应用日志装饰：

@app.get("/")
@LogEndpointAction(entity_name='Sales', action='Get all')
async def root(request: Request):
    return {"message": "Hello World"}

2. 优化Request对象获取与处理 #

在端点函数中显式注入Request参数，同时在装饰器中增加空值判断和异常捕获，避免请求解析失败：
修改装饰器的wrapper部分：

@wraps(func)
async def wrapper(*args, **kwargs):
    request: Request = kwargs.get('request')
    if not request:
        return await func(*args, **kwargs)
    
    current_user = await self._authenticate_user(request)
    response = await func(*args, **kwargs)
    
    # 安全解析请求体
    body = {}
    if request.method in ["POST", "PUT", "PATCH"]:
        try:
            body = await request.json()
        except Exception:
            body = {}

    log_data = {
        'tenant_id': '',
        'user_id': '',
        'full_name': '',
        'roles': [],
        'email': '',
        'entity_name': self.entity_name,
        'data': json.dumps(body),
        'i18n_key': self.action
    }

    await self.db.execute(query=INSERT_LOG_QUERY, values=log_data)
    return response

补充说明 #

如果不想在每个端点手动注入Request参数，可以通过Starlette的请求上下文获取，但显式注入的方式更直观且不易出错。

内容的提问来源于stack exchange，提问作者Lutaaya Huzaifah Idris