FastAPI + SQLAlchemy 实战：从零构建高效父-子关系API

大家好，今天我要分享一个实用的FastAPI开发案例，带你从零开始构建一个包含父子关系的数据模型API。无论你是FastAPI新手还是想深入了解SQLAlchemy ORM的使用，这篇文章都能给你带来收获。

为什么选择FastAPI？

FastAPI作为Python领域最快的Web框架之一，凭借其自动生成API文档、基于Python类型提示、高性能等特性，正在成为越来越多开发者的首选。结合SQLAlchemy这个强大的ORM工具，我们可以快速构建出规范、高效的RESTful API。

项目背景

我们将构建一个简单的”父母-孩子”关系管理系统。这个场景很常见，比如：

• 电商中的订单-订单项关系
• 博客中的文章-评论关系
• 组织架构中的部门-员工关系

通过这个案例，你将学会如何处理这种典型的一对多关系。

环境准备

首先安装必要的依赖：

pip install fastapi sqlalchemy uvicorn pydantic

代码实现详解

1. 数据库配置

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 使用SQLite数据库，方便快速开发
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False}  # SQLite特定配置
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

关键点：

• check_same_thread=False 允许在不同线程中使用同一个SQLite连接
• sessionmaker 创建数据库会话工厂
• declarative_base() 创建模型基类

2. 数据库依赖注入

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

这个生成器函数是FastAPI依赖注入系统的核心，确保每个请求都有独立的数据库会话，请求结束后自动关闭连接。

3. SQLAlchemy模型设计

class ParentDB(Base):
    __tablename__ = "parents"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    # 建立一对多关系
    children = relationship("ChildDB", back_populates="parent")

class ChildDB(Base):
    __tablename__ = "children"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    parent_id = Column(Integer, ForeignKey("parents.id"))
    parent = relationship("ParentDB", back_populates="children")

设计亮点：

• 使用ForeignKey建立外键约束
• relationship创建ORM级别的双向关系
• back_populates明确指定对方的关系属性

4. Pydantic模型（Schema）

class Child(BaseModel):
    name: str

    class Config:
        orm_mode = True

class Parent(BaseModel):
    name: str
    children: list[Child] = []  # 默认空列表

    class Config:
        orm_mode = True

Pydantic模型的作用：

• 请求体验证
• 响应数据序列化
• orm_mode=True 允许从ORM模型自动转换

5. API端点实现

创建父母

@app.post("/parents/", response_model=Parent)
def create_parent(parent: Parent, db: Session = Depends(get_db)):
    db_parent = ParentDB(name=parent.name)
    db.add(db_parent)
    db.commit()
    db.refresh(db_parent)
    return db_parent

为孩子添加父母

@app.post("/parents/{parent_id}/children/", response_model=Child)
def create_child(parent_id: int, child: Child, db: Session = Depends(get_db)):
    db_child = ChildDB(name=child.name, parent_id=parent_id)
    db.add(db_child)
    db.commit()
    db.refresh(db_child)
    return db_child

查询父母及其孩子

@app.get("/parents/{parent_id}/", response_model=Parent)
def read_parent(parent_id: int, db: Session = Depends(get_db)):
    db_parent = db.query(ParentDB).filter(ParentDB.id == parent_id).first()
    if db_parent is None:
        raise HTTPException(status_code=404, detail="Parent not found")
    return db_parent

注意：由于我们配置了relationship，当返回db_parent时，SQLAlchemy会自动加载关联的children数据。

更新父母信息

@app.put("/parents/{parent_id}/", response_model=Parent)
def update_parent(parent_id: int, parent: Parent, db: Session = Depends(get_db)):
    db_parent = db.query(ParentDB).filter(ParentDB.id == parent_id).first()
    if db_parent is None:
        raise HTTPException(status_code=404, detail="Parent not found")
    db_parent.name = parent.name
    db.commit()
    db.refresh(db_parent)
    return db_parent

删除父母

@app.delete("/parents/{parent_id}/")
def delete_parent(parent_id: int, db: Session = Depends(get_db)):
    db_parent = db.query(ParentDB).filter(ParentDB.id == parent_id).first()
    if db_parent is None:
        raise HTTPException(status_code=404, detail="Parent not found")
    db.delete(db_parent)
    db.commit()
    return {"message": "Parent deleted successfully"}

级联删除说明：默认情况下，删除Parent时，如果有关联的Child，数据库会报错。实际项目中可以根据需求配置cascade="all, delete"参数。

运行测试

创建文件main.py，复制以上代码，然后运行：

uvicorn main:app --reload

访问 http://localhost:8000/docs 查看自动生成的交互式API文档。

API测试流程

1. 创建父母：POST /parents/ {“name”: “张三”}
2. 添加孩子：POST /parents/1/children/ {“name”: “张小三”}
3. 查询父母：GET /parents/1/（会同时返回孩子信息）
4. 更新父母：PUT /parents/1/ {“name”: “张四”}
5. 删除父母：DELETE /parents/1/

进阶优化建议

1. 添加数据验证：在Pydantic模型中添加字段验证

from pydantic import Field

class Child(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)

2. 处理级联操作：在relationship中配置cascade

children = relationship("ChildDB", back_populates="parent", cascade="all, delete-orphan")

3. 添加分页功能：为列表接口添加offset/limit参数

4. 使用异步数据库驱动：配合databases或SQLAlchemy 1.4+的异步特性

总结

通过这个实战案例，我们学习了：

• FastAPI的项目结构和依赖注入
• SQLAlchemy ORM的基本使用
• 一对多关系的建模方法
• RESTful API的设计规范
• Pydantic模型的验证和序列化

这个基础模板可以轻松扩展到更复杂的业务场景。FastAPI + SQLAlchemy的组合，既能保证开发效率，又能提供良好的性能和代码可维护性。

希望这篇文章对你有帮助！如果你在实践过程中遇到任何问题，欢迎在评论区留言讨论。