用FastAPI搭建用户管理微服务需四步:定义对齐的ORM模型与Pydantic Schema;用Depends注入数据库会话;实现带查重、加密、刷新的健壮CRUD路由;启动时通过startup事件或Alembic初始化数据库表。

用 FastAPI 搭建微服务,关键不是堆功能,而是把接口、数据、校验、错误处理这几块稳稳地串起来。下面以一个“用户管理微服务”为例,手把手整合数据库(PostgreSQL + SQLAlchemy ORM),不绕弯、不炫技,只讲实际开发中必须踩准的点。
定义清晰的数据模型与 Pydantic Schema
别一上来就写 API,先理清“用户”在数据库里长什么样、对外暴露哪些字段、哪些必填、哪些可选。FastAPI 靠 Pydantic 做自动校验和文档生成,这一步定调后续所有交互。
- 用 SQLAlchemy Base 定义 ORM 模型(含主键、索引、默认值)
- 用 Pydantic BaseModel 定义请求体(
UserCreate)、响应体(UserOut)、更新体(UserUpdate),字段类型与数据库对齐,但可按需裁剪(比如密码不返回) - 注意:数据库模型用
id: int,响应模型用id: int | None = None—— 因为新建时 ID 是数据库自增,响应才带值
用依赖注入管理数据库会话
FastAPI 的 Depends 不是装饰器噱头,是解耦核心。数据库连接不能在每个路由里手动 create_engine,而应封装成可复用、可测试、带生命周期管理的依赖。
- 写一个
get_db()依赖函数,内部用SessionLocal()获取会话,用try/finally确保session.close() - 在路由函数参数中声明
db: Session = Depends(get_db),FastAPI 自动注入并回收 - 开发时加
echo=True查 SQL;上线关掉,避免日志爆炸
CRUD 路由要兼顾简洁与健壮
微服务接口不是越短越好,而是错误能明确报、业务逻辑不裸奔、边界条件有兜底。例如创建用户:
立即学习“Python免费学习笔记(深入)”;
- 接收
UserCreate,用db.query(User).filter(User.email == user_in.email).first()查重,别靠数据库唯一约束硬扛——前端需要友好提示 - 密码用
pwdlib.hash()加密后再存,绝不存明文 - 成功后用
db.refresh(user)刷新对象,确保 ID、created_at 等数据库生成字段同步到响应中 - 删除接口别直接
db.delete(),先get再删,查不到就抛HTTPException(status_code=404)
启动服务前务必初始化数据库表
SQLAlchemy 不会自动建表(除非你用 Base.metadata.create_all() 主动触发)。微服务启动时若表不存在,第一次查询就崩。
- 写一个
init_db(db: Session)函数,在 FastAPI 的startup事件中调用 - 或更稳妥:用
alembic做迁移管理,alembic revision --autogenerate -m "init"+alembic upgrade head - 本地开发可用
drop_all() + create_all()快速重置,但禁止在生产环境用
跑通这四步,你就有了一个可运行、可调试、可扩展的 FastAPI 微服务骨架。后续加 JWT 认证、Redis 缓存、异步任务,都是在这个结构上自然叠加,而不是推倒重来。










