简介：本资源是一个基于FastAPI-Users的轻量级用户管理系统实战示例，面向Python后端开发者及FastAPI初学者，解决RESTful API场景下快速集成安全认证与权限管理的共性难题。压缩包共11个文件，含6个核心Python源码（如main.py、users.py、db.py、config.py等）与5个编译缓存文件（pyc），总大小仅11KB，结构精简，聚焦数据库连接配置、用户模型定义、FastAPI-Users初始化及路由集成等关键环节，便于快速理解框架集成逻辑与最小可行实践路径。已有483人学习下载，适合希望避开重复造轮子、直接复用成熟认证方案的中初级开发者。读者可直接运行调试，掌握OAuth2.0密码流认证、JWT令牌签发验证、用户状态管理及SQLAlchemy基础集成等核心能力，并基于此模板扩展角色权限、邮箱激活等业务功能。
1. 为什么用 FastAPI-Users 而不是手写 JWT 中间件？——一个真实压测场景下的取舍
上周上线一个内部数据看板 API，初期用自研的JWTBearer+ SQLAlchemy 用户表硬扛，结果在并发 300+ 时/me接口平均延迟跳到 850ms，排查发现 62% 的耗时卡在密码校验（bcrypt.verify同步阻塞）和重复的user = db.query(User).filter(...).first()查询上。换成 FastAPI-Users 后，同样负载下延迟稳定在 92ms，关键不是它“封装了什么”，而是它把认证路径上的每一步都做了异步化、缓存化、可插拔化：密码校验走asyncpg兼容的passlib异步后端，用户查询自动带selectinload预加载角色关系，JWT 签发/验证全程不碰数据库。这不是“省事”，是把用户管理从“业务附属品”变成可独立压测、可灰度发布、可按需替换认证源（比如下周要接入企业微信 OAuth2）的模块。适合所有正在用 FastAPI 构建中后台系统、且用户量预期会突破 10k 的团队——尤其当你发现auth.py文件已超 800 行、requirements.txt里pydantic和passlib版本开始打架时。
2. 从 models.py 到 main.py：FastAPI-Users 的四层依赖注入链
FastAPI-Users 不是“加个 router 就完事”的库，它的核心价值藏在类型安全的依赖注入链里。整个流程像一条精密流水线：数据库模型 → 用户管理器 → 认证后端 → FastAPI-Users 实例。漏掉任意一环，要么启动报错，要么认证逻辑失效。下面以你下载包里的models/users.py为蓝本，逐层拆解这个链条如何咬合。
2.1 用户模型必须继承 BaseUser 和 BaseUserCreate：为什么 Pydantic 模型不能直接用？
FastAPI-Users 要求用户模型严格遵循其协议，不是随便定义个class User(BaseModel)就行。你包里的models/users.py里这段代码是起点：
# models/users.py from fastapi_users import schemas from sqlalchemy import Boolean, Integer, String from sqlalchemy.orm import Mapped, mapped_column from db import Base class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True) email: Mapped[str] = mapped_column(String, unique=True, index=True) hashed_password: Mapped[str] = mapped_column(String) is_active: Mapped[bool] = mapped_column(Boolean, default=True) is_superuser: Mapped[bool] = mapped_column(Boolean, default=False) is_verified: Mapped[bool] = mapped_column(Boolean, default=False)但仅此不够。FastAPI-Users 需要两套 Pydantic 模型来约束输入输出格式，它们必须继承特定基类：
# models/__init__.py from fastapi_users import schemas from pydantic import EmailStr class UserRead(schemas.BaseUser[int]): id: int email: EmailStr is_active: bool is_superuser: bool is_verified: bool class UserCreate(schemas.BaseUserCreate): email: EmailStr password: str class UserUpdate(schemas.BaseUserUpdate): password: str | None = None email: EmailStr | None = None is_active: bool | None = None is_superuser: bool | None = None is_verified: bool | None = None注意：
BaseUser[int]的泛型参数int必须与User.id的类型一致（这里是Mapped[int]），否则 FastAPI-Users 在生成 OpenAPI 文档时会报TypeError: Type 'int' is not valid。很多新手卡在这里，以为是数据库配置问题，其实是泛型没对齐。
2.2 用户管理器 UserManager：把密码哈希、邮箱验证、状态检查全收口
UserManager是 FastAPI-Users 的心脏，它把所有业务逻辑（密码重置、邮箱验证、封禁用户）封装成方法，并强制注入数据库会话。你包里的models/users.py应该有类似实现：
# models/users.py from fastapi_users.manager import BaseUserManager, UUIDIDMixin from fastapi_users import exceptions from typing import Optional, TYPE_CHECKING if TYPE_CHECKING: from models import User # noqa: F401 class UserManager(BaseUserManager["User", int]): reset_password_token_secret = "SECRET_RESET" verification_token_secret = "SECRET_VERIFY" async def on_after_register(self, user: "User", request: Optional[Request] = None): print(f"User {user.id} has registered.") async def on_after_forgot_password(self, user: "User", token: str, request: Optional[Request] = None): print(f"User {user.id} has forgot their password. Reset token: {token}") async def validate_password(self, password: str, user: "User") -> None: if len(password) < 8: raise exceptions.InvalidPasswordException( reason="Password should be at least 8 characters" )关键点在于BaseUserManager["User", int]的泛型声明：第一个参数是你的 SQLAlchemy 模型类名（字符串引用避免循环导入），第二个是主键类型。这个类会自动获得create,get_by_email,update等方法，但所有方法都要求传入AsyncSession——这正是db.py里get_async_session的用武之地。
2.3 认证后端：JWT vs DatabaseToken，选哪个取决于你的部署形态
FastAPI-Users 支持多种认证方式，但实际项目中 90% 用的是JWTStrategy。你包里的config.py很可能已配置好：
# config.py from fastapi_users.authentication import JWTStrategy, AuthenticationBackend from fastapi_users import FastAPIUsers from models.users import User from models import get_user_manager from db import get_async_session SECRET = "SECRET_JWT" def get_jwt_strategy() -> JWTStrategy: return JWTStrategy(secret=SECRET, lifetime_seconds=3600) auth_backend = AuthenticationBackend( name="jwt", transport=CookieTransport(cookie_max_age=3600), # 或 BearerTransport() get_strategy=get_jwt_strategy, )这里有两个易错点：
lifetime_seconds=3600是 JWT 过期时间，单位秒。若前端需要长登录，别盲目调大，应配合refresh_token流程（FastAPI-Users 本身不提供 refresh token，需自行扩展）。CookieTransport和BearerTransport的选择决定前端如何传 token：前者走Set-Cookie头，后者走Authorization: Bearer <token>。你包里main.py的fastapi_users.get_login_router(auth_backend)会根据 transport 自动适配路由。
2.4 FastAPIUsers 实例：把前三层组装成可挂载的路由集合
最后一步，把模型、管理器、后端三者注入FastAPIUsers：
# main.py from fastapi import FastAPI from models import User, UserRead, UserCreate, UserUpdate from models.users import get_user_manager from config import auth_backend from fastapi_users import FastAPIUsers fastapi_users = FastAPIUsers[User, int]( get_user_manager, [auth_backend], ) app = FastAPI() # 挂载路由 app.include_router( fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"], ) app.include_router( fastapi_users.get_register_router(UserRead, UserCreate), prefix="/auth", tags=["auth"], ) app.include_router( fastapi_users.get_reset_password_router(), prefix="/auth", tags=["auth"], ) app.include_router( fastapi_users.get_verify_router(UserRead), prefix="/auth", tags=["auth"], ) app.include_router( fastapi_users.get_users_router(UserRead, UserUpdate), prefix="/users", tags=["users"], )提示：
FastAPIUsers[User, int]的泛型必须与UserManager一致。如果此处写成FastAPIUsers[User, str]，启动时会报TypeError: Cannot resolve type argument，因为User.id是int类型。
3. 数据库集成实战：SQLAlchemy 1.4+ 异步会话的三处关键配置
FastAPI-Users 默认支持异步数据库操作，但你的db.py若沿用旧式同步连接，会导致整个认证链路阻塞。你下载包里的db.py和db_connect.py需要按以下方式重构，否则UserManager.create()会抛RuntimeError: There is no current event loop in thread。
3.1 创建异步引擎：不要用 create_engine，要用 create_async_engine
# db.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import declarative_base from sqlalchemy.ext.asyncio import async_sessionmaker DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname" engine = create_async_engine( DATABASE_URL, echo=True, # 开发时开启，生产环境设为 False pool_pre_ping=True, # 连接前检测是否存活 pool_recycle=3600, # 连接复用 1 小时 ) async_session = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) Base = declarative_base()注意：
postgresql+asyncpg://是必须的驱动前缀，asyncpg比psycopg2性能高 30% 以上。若用 SQLite，需改用aiosqlite驱动，且pool_pre_ping不可用。
3.2 依赖注入：get_async_session 必须返回 AsyncSession，且带 @contextlib.asynccontextmanager
# db.py from contextlib import asynccontextmanager from typing import AsyncGenerator @asynccontextmanager async def get_async_session() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: yield session这个@asynccontextmanager是关键——它让 FastAPI 能在请求生命周期内正确管理会话的创建和关闭。如果你的db_connect.py里还是def get_db():这种同步写法，必须重写。
3.3 在 UserManager 中注入会话：用 Depends(get_async_session) 替代手动创建
# models/users.py from fastapi import Depends from db import get_async_session async def get_user_manager( user_db: SQLAlchemyUserDatabase = Depends(get_user_db), ) -> UserManager: return UserManager(user_db)而get_user_db的实现必须是：
# models/__init__.py from fastapi_users.db import SQLAlchemyUserDatabase from db import get_async_session from models.users import User async def get_user_db( session: AsyncSession = Depends(get_async_session), ) -> SQLAlchemyUserDatabase: yield SQLAlchemyUserDatabase(session, User)这里yield而非return，是因为SQLAlchemyUserDatabase需要会话在整个请求周期内保持活跃。若写成return，会话会在get_user_db返回时被关闭，后续UserManager.create()就会报ObjectDisposedError。
3.4 初始化数据库表：用 alembic 还是 Base.metadata.create_all？
FastAPI-Users 不提供数据库迁移工具，推荐用 Alembic。但开发阶段快速验证，可临时用Base.metadata.create_all：
# db.py (末尾添加) async def init_db(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) # 在 main.py 的 lifespan 中调用 @app.on_event("startup") async def startup(): await init_db()警告：
create_all不会处理字段变更（如新增is_verified字段），生产环境必须用 Alembic 生成 migration 脚本。你包里的__pycache__目录下没有.pyc文件，说明还没跑过alembic revision --autogenerate。
4. 权限控制进阶：用 get_required_current_user 替代 get_current_user 实现 RBAC
FastAPI-Users 默认的get_current_user只做身份校验，不检查权限。要实现角色权限控制（RBAC），必须自定义依赖项。你包里的main.py可能只挂了基础路由，现在补上管理员专属接口：
4.1 定义角色枚举和权限检查函数
# models/__init__.py from enum import Enum from fastapi import Depends, HTTPException, status from fastapi_users import models from fastapi_users.manager import BaseUserManager class Role(str, Enum): USER = "user" ADMIN = "admin" SUPERUSER = "superuser" async def get_required_current_user( user: models.UP = Depends(fastapi_users.current_user()), ) -> models.UP: if not user.is_active: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user", ) return user async def get_admin_user( user: models.UP = Depends(get_required_current_user), ) -> models.UP: if not user.is_superuser: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions", ) return user4.2 在路由中使用权限依赖项
# main.py @app.get("/admin/users", tags=["admin"]) async def get_all_users( user: User = Depends(get_admin_user), session: AsyncSession = Depends(get_async_session), ) -> List[UserRead]: result = await session.execute(select(User)) users = result.scalars().all() return [UserRead.from_orm(u) for u in users]4.3 扩展用户模型：在 User 表中增加 role 字段并同步更新 Pydantic 模型
# models/users.py class User(Base): # ... 原有字段 role: Mapped[str] = mapped_column(String, default=Role.USER.value) # models/__init__.py class UserRead(schemas.BaseUser[int]): # ... 原有字段 role: Role这样，当管理员调用/admin/users时，FastAPI-Users 会先执行get_required_current_user（检查激活状态），再执行get_admin_user（检查is_superuser），双重保障。比在每个路由里写if not user.is_superuser: raise ...更符合依赖注入原则。
5. 生产环境必调参数：JWT 密钥轮换、密码策略、邮箱验证超时的实操值
FastAPI-Users 提供了大量可调参数，但文档很少说明“为什么设这个值”。以下是经过 3 个线上项目验证的生产级配置，直接抄作业：
| 参数 | 推荐值 | 为什么这么设 | 对应代码位置 |
|---|---|---|---|
JWTStrategy.lifetime_seconds | 1800(30分钟) | 防止 token 泄露后长期有效；前端应实现自动刷新逻辑 | config.py |
UserManager.reset_password_token_secret | 32字节随机密钥（secrets.token_urlsafe(32)） | 避免重置链接被预测；每次部署生成新密钥 | models/users.py |
UserManager.verification_token_secret | 独立于 reset 的密钥 | 邮箱验证和密码重置密钥分离，降低单点泄露风险 | models/users.py |
SQLAlchemyUserDatabase的session | 设置expire_on_commit=False | 避免 commit 后对象属性变None，导致UserRead.from_orm(user)报错 | db.py |
CookieTransport.cookie_secure | True（HTTPS 环境） | 强制 cookie 只在 HTTPS 下发送，防止中间人窃取 | config.py |
验证这些配置是否生效，最直接的方法是抓包测试：
# 1. 注册用户，观察响应头 Set-Cookie 是否含 Secure 属性 curl -X POST http://localhost:8000/auth/register \ -H "Content-Type: application/json" \ -d '{"email":"test@example.com","password":"Passw0rd!"}' # 2. 登录后，用返回的 cookie 访问 /me，确认返回 200 curl -X GET http://localhost:8000/users/me \ -H "Cookie: fastapiusersauth=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." \ -v若Set-Cookie中没有Secure，检查config.py里CookieTransport的cookie_secure是否为True，且 Nginx/Apache 反向代理已正确设置X-Forwarded-Proto: https。这是线上环境最常见的 401 错误根源——不是代码问题，是反向代理没透传协议头。
FastAPI-Users 的get_current_user依赖项在request.state.user中缓存用户对象，因此同一请求内多次调用不会重复查询数据库。你可以通过在on_after_login回调中打印id(request.state.user)来验证是否为同一实例。
本文还有配套的精品资源，点击获取
