构建模块化的FastAPI应用: 从用户认证到角色控制

实现了用户身份验证及角色授权的基本功能。具体来说，当用户尝试访问某些资源时，系统会首先验证用户的身份，然后根据用户的角色来决定是否允许访问特定资源。例如，普通用户只能访问自己的信息，而管理员可以访问额外的管理界面。这种机制保证了系统的安全性，并且可以根据需要灵活地扩展不同的角色和权限。
在这里插入图片描述

用户模型定义：定义了一个用户模型 User，包含用户名和角色信息。此外还定义了一个 UserInDB 模型，包含用户密码，用于表示数据库中的用户信息。
用户身份验证：实现了一个依赖项 get_current_user，通过请求头部的 token 字段来确定当前用户的身份。如果 token 对应的用户存在于模拟数据库 FAKE_USERS_DB 中，则返回该用户对象；否则抛出异常，提示用户不存在或 Token 无效。
管理员权限检查：定义了一个依赖项 get_admin_user，基于 get_current_user 函数来进一步确认当前用户是否具有管理员权限。如果不是管理员，则抛出异常，提示需要管理员权限。
用户相关API：在 users 路由模块中定义了几个与用户相关的路径操作。这些操作都需要通过 get_current_user 依赖项来验证当前用户的合法性。例如，read_users 方法返回当前用户的列表，read_user_me 方法返回当前用户的信息，而 read_user 方法则返回指定用户名对应的用户信息，前提是当前用户与请求的用户名匹配。
管理员相关API：在 admin 路由模块中定义了管理员专属的路径操作。这些操作需要通过 get_admin_user 依赖项来确保只有管理员可以访问。例如，admin_dashboard 方法返回管理员的信息，只有管理员用户可以调用这个端点。

用户角色验证功能

在 FastAPI 中，通常会使用 Pydantic 模型来定义请求和响应的数据结构。为了实现一个简单的用户角色验证功能，我们可以创建一个模型来表示用户，并包含一个字段来存储用户的角色信息。然后，我们可以在依赖项中使用这个模型来进行角色检查。

首先，我们需要安装必要的依赖：

pip install fastapi pydantic uvicorn

接下来，我们更新文件结构并添加新的模型文件：

更新后的文件结构

.
├── app
│   ├── __init__.py
│   ├── main.py
│   ├── dependencies.py
│   ├── models.py  # 新增模型文件
│   ├── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── users.py
│   └── internal
│       ├── __init__.py
│       └── admin.py

文件内容

app/models.py

用户模型定义。

from pydantic import BaseModelclass User(BaseModel):username: strrole: str  # 角色字段class UserInDB(User):password: str  # 假设数据库中的用户模型包含密码字段

app/dependencies.py

更新依赖项以包含角色检查。

from fastapi import Header, HTTPException, Depends
from .models import User  # 导入用户模型FAKE_USERS_DB = {  # 模拟数据库"johndoe": User(username="johndoe", role="admin"),"alice": User(username="alice", role="user"),"bob": User(username="bob", role="user")
}async def get_current_user(token: str = Header(...)):  # 从标头获取当前用户user = FAKE_USERS_DB.get(token)if not user:raise HTTPException(status_code=400, detail="Token 无效或用户不存在")return userasync def get_admin_user(current_user: User = Depends(get_current_user)):  # 检查是否为管理员if current_user.role != "admin":raise HTTPException(status_code=403, detail="需要管理员权限")return current_user

app/routers/users.py

更新用户路由以使用新依赖。

from fastapi import APIRouter, Depends
from ..dependencies import get_current_user, get_admin_user
from ..models import Userrouter = APIRouter(tags=["users"])@router.get("/users/", response_model=list[User])
async def read_users(current_user: User = Depends(get_current_user)):return [current_user]  # 返回当前用户信息@router.get("/users/me", response_model=User)
async def read_user_me(current_user: User = Depends(get_current_user)):return current_user@router.get("/users/{username}", response_model=User)
async def read_user(username: str, current_user: User = Depends(get_current_user)):if current_user.username == username:return current_userraise HTTPException(status_code=404, detail="用户未找到")

app/routers/admin.py

添加管理员路由。

from fastapi import APIRouter, Depends
from ..dependencies import get_admin_user
from ..models import Userrouter = APIRouter(tags=["admin"])@router.get("/admin/", response_model=User)
async def admin_dashboard(current_user: User = Depends(get_admin_user)):return current_user  # 返回当前管理员的信息