一个 put 端点是否有可能在 fastapi、pydantic 上拥有超过 1 种类型的参数

标签：python fastapi pydantic

因此，首先，这个问题的目标是获得更多有关是否可能以及害虫做法的知识。

有 3 种类型的用户 7、6为普通用户 5,4 是出版商 3,2 是经理 1 是 admin

我使用此函数来专用 user_role 这是第一个问题我会经常使用这个函数，我想了一下，

我可以将 user_role 添加到 JWT 令牌，但我认为这会造成安全问题，对吗？

async def check_permission(
    permission: int,
    current_user: user_dependecny,
    db: db_dependecny,
    error: bool = True,
):
    current_user_role = (
        db.query(user.User).filter(user.User.id == current_user.id).first().user_role
    )
    # 7= reader, 6= Pro_reader,
    # 5= publisher, 4= editor,
    # 3= small_manager, 2= big_manager, 1= admin
    if current_user_role <= permission:
        return True
    elif error == False:
        return False
    elif current_user_role > permission:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Do not have premisson."
        )

标题上的主要问题

尝试在一个参数上使用多个 data_type 是可以通过使用 Union

from typing import Annotated, Union, cast
type1 = None
type2 = None
type3 = None
var: Union[type1,type2,type3]

第一次尝试使用 model_response 上的 get enpoint 来实现的，它起作用了。

下面的代码正在做的是根据您的 user_role 返回用户数据并执行此操作我需要 3 个不同的基础模型

@router.get(
    "/{user_id}/", response_model=Union[SelfUserInfo, ManagerUserInfo, UserInfo]
)
async def read_user(
    user_id: int,
    db: db_dependecny,
    current_user: user_dependecny,
):
    db_user = db.query(user.User).filter(user.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="user is not found"
        )
    if await check_permission(3, current_user, db, error=False) == True:
        user_data = ManagerUserInfo(
            username=db_user.username,
            avatar_url=db_user.avatar_url,
            user_role=db_user.user_role,
            email=db_user.email,
            country_id=db_user.country_id,
            last_login=db_user.last_login,
            date_joined=db_user.date_joined,
            is_verfied=db_user.is_verfied,
            state=db_user.state,
            is_staff=db_user.is_staff,
            is_superuser=db_user.is_superuser,
        )
        return user_data
    else:
        if current_user.id == user_id:
            user_data = SelfUserInfo(
                username=db_user.username,
                avatar_url=db_user.avatar_url,
                user_role=db_user.user_role,
                email=db_user.email,
                country_id=db_user.country_id,
                last_login=db_user.last_login,
                date_joined=db_user.date_joined,
                is_verfied=db_user.is_verfied,
            )
            return user_data
        else:
            user_data = UserInfo(
                username=db_user.username,
                avatar_url=db_user.avatar_url,
                user_role=db_user.user_role,
            )
        return user_data

关于此代码的第二个问题是它没有遵循害虫实践。

最后一个代码是我有问题

我的问题是 user_data 类型始终是 SelfUserUpdate

特别是第一个类型|| |如何解决这个问题，这样做可以吗？制作 3 个不同的端点。

user_data: Union[SelfUserUpdate, ManagerUserUpdate, AdminUserUpdate],

how to fix it and is it okey and not a ba practice to do it like that instead of making 3 deferent endpoints.

@router.put("/{user_id}/")
async def Update_user(
    user_id: int,
    current_user: user_dependecny,
    db: db_dependecny,
    user_data: Union[SelfUserUpdate, ManagerUserUpdate, AdminUserUpdate],
):
    db_user = db.query(user.User).filter(user.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="user is not found"
        )
    if await check_permission(1, current_user, db, error=False) == True:
        # Admin edit user
        db_user.username = user_data.username
        db_user.avatar_url = user_data.avatar_url
        db_user.email = user_data.email
        db_user.country_id = user_data.country_id
        db_user.user_role = user_data.user_role
        db_user.state = user_data.state
        db_user.is_staff = user_data.is_staff
        db_user.is_verfied = user_data.is_verfied
        db_user.is_superuser = user_data.is_superuser
        db.commit()
        db.refresh(db_user)
        return user_data

    elif check_permission(3, current_user, db, error=False) == True:
        # Manager edit user
        db_user.username = user_data.username
        db_user.avatar_url = user_data.avatar_url
        db_user.email = user_data.email
        db_user.country_id = user_data.country_id
        db_user.user_role = user_data.user_role
        db_user.state = user_data.state
        db_user.is_staff = user_data.is_staff
        db.commit()
        db.refresh(db_user)
        return user_data
    elif current_user.id == user_id:
        # current user edit him self
        db_user.username = user_data.username
        db_user.avatar_url = user_data.avatar_url
        db_user.email = user_data.email
        db_user.country_id = user_data.country_id
        db.commit()
        db.refresh(db_user)
        return user_data
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="You can't change This user.",
        )

好的，让我们来分解一下如何在你的FastAPI应用程序中处理不同用户角色和数据验证。

1. JWT 与用户角色

你问到在JWT令牌中包含 user_role 是否安全。一般来说，最好不要在JWT令牌本身中包含敏感信息，如角色。这是因为：

JWT令牌未经加密，只是经过签名。 任何拥有该令牌的人都可以对其进行解码并查看其内容。
令牌过期时间可能很长。 如果令牌泄露，它可能会在一段时间内被用于访问敏感信息。

更好的方法：

在用户登录时生成一个JWT令牌，其中仅包含用户ID和其他必要信息。
在每次请求中，使用身份验证中间件从JWT令牌中提取用户ID。
使用用户ID从数据库中查询完整的用户信息，包括角色。

2. 多类型请求体

你想在 PUT /users/{user_id}/ 端点上接受不同的请求体，具体取决于用户角色。虽然使用 Union 在技术上是可行的，但这会导致代码复杂且难以维护。

更好的方法：

为每种用户角色创建单独的Pydantic模型。 这可以更清晰地定义每种角色允许更新哪些字段。
创建一个单一的 UserUpdate 模型，并在其中使用可选字段。 这可以简化路由逻辑，但在处理角色特定验证时需要额外的代码。

以下是使用单独模型的示例：

from typing import Optional
from pydantic import BaseModel

class SelfUserUpdate(BaseModel):
    username: Optional[str] 
    avatar_url: Optional[str] 
    email: Optional[str]
    country_id: Optional[int]

class ManagerUserUpdate(BaseModel):
    username: Optional[str] 
    avatar_url: Optional[str] 
    email: Optional[str]
    country_id: Optional[int]
    user_role: Optional[int]
    state: Optional[str]
    is_staff: Optional[bool]

class AdminUserUpdate(BaseModel):
    # 所有字段都是可选的，因为管理员可以更新任何内容
    username: Optional[str] 
    avatar_url: Optional[str] 
    email: Optional[str] 
    country_id: Optional[int]
    user_role: Optional[int]
    state: Optional[str]
    is_staff: Optional[bool]
    is_verified: Optional[bool]
    is_superuser: Optional[bool]

@router.put("/{user_id}/")
async def update_user(
    user_id: int,
    current_user: user_dependecny,
    db: db_dependecny,
    user_data: Union[SelfUserUpdate, ManagerUserUpdate, AdminUserUpdate] = Body(...),
):
    db_user = db.query(user.User).filter(user.User.id == user_id).first()

    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if current_user.id == user_id:
        # 自我更新
        if isinstance(user_data, SelfUserUpdate):
            # 应用更新
            update_data = user_data.dict(exclude_unset=True)
            for key, value in update_data.items():
                setattr(db_user, key, value)
        else:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions"
            )

    elif await check_permission(3, current_user, db, error=False):
        # 管理员更新
        if isinstance(user_data, (ManagerUserUpdate, AdminUserUpdate)):
            # 应用更新
            update_data = user_data.dict(exclude_unset=True)
            for key, value in update_data.items():
                setattr(db_user, key, value)
        else:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions"
            )

    elif await check_permission(1, current_user, db, error=False):
        # 超级管理员更新
        if isinstance(user_data, AdminUserUpdate):
            # 应用更新
            update_data = user_data.dict(exclude_unset=True)
            for key, value in update_data.items():
                setattr(db_user, key, value)
        else:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions"
            )
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions"
        )

    db.commit()
    db.refresh(db_user)
    return user_data

3. 代码改进建议

不要在 check_permission 函数中使用 error 参数。 直接引发异常会更清晰，并且可以让你利用FastAPI的自动异常处理机制。
使用Pydantic模型来验证数据库查询结果。 这可以确保数据一致性，并简化代码。
考虑使用依赖注入来简化数据库访问。

通过遵循这些建议，你可以创建一个更安全、更易于维护和更符合RESTful API最佳实践的FastAPI应用程序。

标签：python,fastapi,pydantic
From： 78790857

一个 put 端点是否有可能在 fastapi、pydantic 上拥有超过 1 种类型的参数

相关文章

赞助商

阅读排行