FastAPI & SQLAlchemy 实战（十）：FastAPI + SQLAlchemy，一个可运行的增删改查接口

「FastAPI & SQLAlchemy 实战」系列共 10 篇，从协程基础讲到数据库接口。这是最后一篇。

结论：用 Depends(get_db) 给每个请求注入一个 Session，接口里只做业务逻辑；请求结束自动关闭会话。这样就把前面学的 FastAPI 和 SQLAlchemy 接成了一套可运行的服务。

一、项目结构

myproject/
├── main.py            # FastAPI 接口
├── base.py            # Base 基类
├── models.py          # ORM 模型
└── database.py        # Engine + SessionLocal

二、依赖：每个请求一个 Session

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import SessionLocal, engine
from models import Department, Employee

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

yield 之前是准备，finally 是收尾。FastAPI 保证请求结束后关闭会话，即使接口抛异常也会执行。

三、接口：增、查、按 ID 查

@app.post("/departments/")
def create_department(name: str, location: str, db: Session = Depends(get_db)):
    department = Department(name=name, location=location)
    db.add(department)
    db.commit()
    db.refresh(department)
    return department

@app.get("/departments/")
def list_departments(db: Session = Depends(get_db)):
    return db.query(Department).all()

@app.get("/departments/{department_id}")
def get_department(department_id: int, db: Session = Depends(get_db)):
    department = db.query(Department).filter(Department.id == department_id).first()
    if not department:
        raise HTTPException(status_code=404, detail="Department not found")
    return department

Depends(get_db) 写在参数里，FastAPI 会在调用接口前创建 Session，接口结束后关闭。业务函数不需要关心连接管理。

四、可选：从现有表生成模型

如果数据库里已经有表，可以用 sqlacodegen 生成模型，省去手写：

pip install sqlacodegen pymysql
python -m sqlacodegen mysql+pymysql://root:123456@localhost:3306/fastapi_db > models_generated.py

生成的模型适合做起点，字段注释、关系名、级联策略仍要按业务调整。

五、验证

uvicorn main:app --reload

# 新增
curl -s -X POST "http://127.0.0.1:8000/departments/?name=研发部&location=北京"

# 列表
curl -s http://127.0.0.1:8000/departments/

# 按 ID
curl -s http://127.0.0.1:8000/departments/1
curl -s http://127.0.0.1:8000/departments/999   # 404

打开 /docs 可以直接在页面上点“Try it out”，不用手写 curl。

六、常见错误

• 把 Session 做成全局单例：并发请求会互相污染，必须每请求一个；
• 忘记 commit：数据没写进库；
• 忘记 refresh：拿不到自增 ID；
• 查不到直接返回 None：前端拿到 200 空数据，应该返回 404；
• 在异步路由里用同步 SQLAlchemy：会阻塞事件循环，改用异步引擎或把接口写成 def。

系列小结

10 篇走下来，主线是：协程解决 I/O 并发，ASGI 让 FastAPI 能用协程，Pydantic 负责参数校验，SQLAlchemy 负责数据库，最后用 Depends(get_db) 把它们接起来。

这套组合适合中小型 API：代码量少、类型清晰、自动文档、数据库操作可控。下一步可以补迁移（Alembic）、鉴权、分页和测试。

上一篇：SQLAlchemy CRUD、关联关系与加载策略