带有 FastAPI 端点参数的装饰器答案

【问题标题】：Decorator WITH arguments for FastAPI endpoint带有 FastAPI 端点参数的装饰器
【发布时间】：2023-02-09 03:36:04
【问题描述】：

我有这个装饰器：

def security(required_roles):
    def decorator(function):
        async def wrapper():
            print("ROLES", required_roles)
            return function
        return wrapper
    return decorator

这个端点，我想装饰：

@app.get(
    "/me", summary="Get details of currently logged in user", response_model=SystemUser
)
@security(required_roles=["role1", "role2"])
async def get_me(user: SystemUser = Depends(get_current_user)):
    return user

但是当我调用它时，我得到了这个：

File "/home/niels/PycharmProjects/fastApiProject/venv/lib/python3.10/site-packages/fastapi/routing.py", line 139, in serialize_response
    raise ValidationError(errors, field.type_)
pydantic.error_wrappers.ValidationError: 1 validation error for SystemUser
response
  value is not a valid dict (type=type_error.dict)

谁能告诉我为什么以及如何重写装饰器。如果我放置装饰器前@app.get(...) 它没有被执行，也不知道为什么。任何帮助将非常感激。

【问题讨论】：

标签： python fastapi python-decorators

【解决方案1】：

您应该将包装器参数传递给包装函数和await它：

def security(required_roles):
    def decorator(function):
        async def wrapper(user):
            print("ROLES", required_roles)
            return await function(user)
        return wrapper
    return decorator

【讨论】：

然后我得到：{"detail": [{"loc": ["query", "args"], "msg": "field required", "type": "value_error.missing" }, { "loc": ["query", "kwargs"], "msg": "field required", "type": "value_error.missing"}]}
如果我将 *args, **kwargs 移动到 decorator(...)，我得到：{"dependency": {},use_cache": true} 显然，由于某种原因，Depends 不会被执行。当我打印它时，我只看到Depends(get_current_user)
FastAPI 检查函数的参数名称作为 GET 查询的参数名称，因此包装器需要与包装函数具有相同的参数。然后相应地更新了我的答案。

【解决方案2】：

我通过执行以下操作让我的装饰器工作：

装饰师：

def has_permission(permission: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            api_key = kwargs.get("api_key_header")
            user_service = kwargs.get("user_service")

            user = user_service.get_by_api_key(api_key)

            if permission not in user.permissions
                raise HTTPException(status_code=403, detail="User doesn't have required permissions")

            return await func(*args, **kwargs)

        return wrapper

    return decorator

路线：

@router.get("/", summary="Retrieve all users.", response_model=List[UserResponse])
@has_permission("users.get")
async def get_all_users(
    api_key_header: str = Security(api_key_header), user_service: UserService = Depends(UserService)
):
    """
    Retrieve all users.
    """
    return user_service.get_all_users()

不是我如何通过kwargs.get()访问路由参数。

【讨论】：