FastAPI 重要知识点——数据库表间的连接操作

在 Web 应用开发中，很少有数据是孤立存在的。比如一个博客系统里，一篇文章下面有很多条评论，一个用户可能发表过多篇文章——这些数据之间天然存在着千丝万缕的联系。今天这篇博客，我们就来彻底搞懂：如何在 FastAPI 中优雅地处理数据库表之间的关联关系。

---

一、先理解关系数据库的核心概念

在动手写代码之前，我们需要先搞懂两个核心概念：外键 和 关系类型。

1.1 什么是外键（Foreign Key）？

外键就是一个表中的字段，它引用了另一个表中的主键字段。 外键的主要作用是保证数据的完整性和关联性，确保一个表中的某个字段的值必须存在于另一个表的对应字段中。

打个比方：你手里有一张员工卡，上面印着你的工号。这个工号在公司的员工档案系统里是唯一的（这就是“主键”）。而你的考勤记录表上，每一条打卡记录都会记录一个“员工工号”字段，这个字段就指向了员工档案表的主键——这就是“外键”。

外键约束的作用：

• 数据完整性：防止插入无效的引用数据（比如考勤记录里不能出现一个不存在的工号）
• 关系映射：实现表与表之间的关联，方便通过 ORM 进行数据查询

1.2 三种常见的表关系

在关系型数据库中，表与表之间的关系主要有三种类型：

关系类型	说明	生活例子
一对多（1:N）	一条主表记录可以关联多条从表记录	一个用户 → 多篇文章；一个班级 → 多名学生
多对一（N:1）	多条从表记录关联到同一条主表记录	多篇文章 → 同一个作者
多对多（N:M）	多条记录互相交叉关联	一个学生选多门课，一门课有多名学生

小提示：一对多和多对一本质上是同一种关系，只是从不同角度观察而已。从用户看文章是一对多，从文章看用户是多对一。

---

二、在 SQLAlchemy 中定义一对多关系

FastAPI 官方推荐使用 SQLAlchemy 作为 ORM（对象关系映射）工具，它能把数据库表“映射”成 Python 类，让我们用面向对象的方式来操作数据。

2.1 模型定义：User 和 Post

假设我们要做一个简单的博客系统，一个用户可以发表多篇文章。这就是典型的一对多关系。

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
from typing import List, Optional

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, nullable=False)
    email = Column(String, unique=True, nullable=False)
    
    # 一对多关系：一个用户有多篇文章
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    
    # 外键：指向 users 表的 id 字段
    user_id = Column(Integer, ForeignKey("users.id"))
    
    # 多对一关系：一篇文章属于一个作者
    author = relationship("User", back_populates="posts")

2.2 关键代码解读

让我们逐行拆解这段代码的核心部分：

（1）定义外键字段

user_id = Column(Integer, ForeignKey("users.id"))

这行代码在 posts 表中创建了一个 user_id 字段，它的值必须来自 users 表的 id 字段。这就是数据库层面的外键约束。

（2）定义 ORM 关系

# 在 User 类中
posts = relationship("Post", back_populates="author")

# 在 Post 类中
author = relationship("User", back_populates="posts")

relationship() 是 SQLAlchemy 提供的关系定义函数，它不在数据库里创建新字段，而是给 Python 类添加一个“虚拟属性”，让你可以通过 user.posts 直接拿到这个用户的所有文章，或者通过 post.author 拿到这篇文章的作者。

（3）back_populates 参数的作用

这个参数告诉 SQLAlchemy：User.posts 和 Post.author 是同一对关系的两端。当你修改一端时，ORM 会自动同步另一端，保持数据一致。

例如：

# 创建一个新用户
new_user = User(username="张三", email="zhangsan@example.com")

# 创建一篇新文章，直接通过关系属性赋值
new_post = Post(title="我的第一篇文章", content="Hello World", author=new_user)

# 此时 new_user.posts 会自动包含这篇新文章
print(new_user.posts)  # 输出: [<Post object>]

这就是 back_populates 的魔力——它让两端关系“双向绑定”。

---

三、在 SQLModel 中定义一对多关系（更简洁的方式）

SQLModel 是 FastAPI 的作者开发的 ORM 库，它结合了 SQLAlchemy 的强大功能和 Pydantic 的数据验证能力，让代码更加简洁。

下面是用 SQLModel 实现同样的 User 和 Post 模型：

from typing import Optional, List
from sqlmodel import Field, SQLModel, Relationship

class User(SQLModel, table=True):
    __tablename__ = "users"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(unique=True, index=True)
    email: str = Field(unique=True, index=True)
    
    # 一对多关系
    posts: List["Post"] = Relationship(back_populates="author")

class Post(SQLModel, table=True):
    __tablename__ = "posts"
    
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    
    # 外键定义
    user_id: Optional[int] = Field(default=None, foreign_key="users.id")
    
    # 多对一关系
    author: Optional[User] = Relationship(back_populates="posts")

对比两种写法，SQLModel 的优势很明显：

• 使用 Python 类型注解代替 Column()，更简洁
• 自动集成了 Pydantic 验证，可以直接用于 API 请求/响应
• Field() 同时支持 ORM 字段定义和数据验证

---

四、在 FastAPI 中实现关联查询的 API

模型定义好了，接下来看看如何在 API 中使用这些关联关系。

4.1 返回嵌套关系的响应

最常见的一个需求：获取文章列表时，同时返回每篇文章的作者信息。

首先，定义 Pydantic 响应模型来控制返回的数据结构：

from pydantic import BaseModel
from typing import Optional, List

# 基础的用户信息
class UserPublic(BaseModel):
    id: int
    username: str
    email: str
    
    class Config:
        from_attributes = True

# 带作者信息的文章响应
class PostWithAuthor(BaseModel):
    id: int
    title: str
    content: str
    author: Optional[UserPublic] = None  # 嵌套作者信息
    
    class Config:
        from_attributes = True

然后编写 API 端点：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import select

app = FastAPI()

@app.get("/posts/", response_model=List[PostWithAuthor])
async def get_posts(db: Session = Depends(get_db)):
    # 使用 select 查询，SQLAlchemy 会自动通过 relationship 加载关联数据
    stmt = select(Post)
    posts = db.execute(stmt).scalars().all()
    return posts

关键在于 Config.from_attributes = True，它告诉 Pydantic 可以从 ORM 对象中读取属性，包括通过 relationship 关联的嵌套对象。

4.2 创建带关联的数据

创建一篇新文章时，需要指定作者：

from fastapi import FastAPI, Depends, HTTPException

class PostCreate(BaseModel):
    title: str
    content: str
    user_id: int

@app.post("/posts/", response_model=PostWithAuthor)
async def create_post(
    post_data: PostCreate,
    db: Session = Depends(get_db)
):
    # 先验证用户是否存在
    user = db.get(User, post_data.user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    # 创建文章，建立关联
    new_post = Post(
        title=post_data.title,
        content=post_data.content,
        user_id=post_data.user_id
    )
    db.add(new_post)
    db.commit()
    db.refresh(new_post)
    return new_post

4.3 获取用户的所有文章

利用 relationship 定义好的 posts 属性，获取用户的所有文章非常直观：

@app.get("/users/{user_id}/posts", response_model=List[PostWithAuthor])
async def get_user_posts(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    # 直接通过关系属性获取，无需手动 JOIN
    return user.posts

user.posts 会自动执行查询，返回该用户的所有文章。这就是 ORM 关系属性的强大之处——你几乎不用写 SQL 就能完成关联查询。

---

五、多对多关系实战

多对多关系比一对多稍微复杂一点，需要一个中间表来存储关联关系。例如：一个学生可以选修多门课程，一门课程也可以有多名学生。

5.1 定义多对多模型

from sqlalchemy import Table, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

# 中间关联表
student_course = Table(
    "student_course",
    Base.metadata,
    Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
    Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True)
)

class Student(Base):
    __tablename__ = "students"
    
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    
    # 多对多关系
    courses = relationship("Course", secondary=student_course, back_populates="students")

class Course(Base):
    __tablename__ = "courses"
    
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    teacher = Column(String)
    
    # 反向关系
    students = relationship("Student", secondary=student_course, back_populates="courses")

secondary=student_course 参数告诉 SQLAlchemy 使用哪个中间表来维护这个多对多关系。

5.2 操作多对多关系

# 给一个学生选课
student = db.get(Student, student_id)
course = db.get(Course, course_id)

# 直接用 append 添加关联
student.courses.append(course)
db.commit()

# 查询学生选了哪些课
student_courses = student.courses

# 查询某门课有哪些学生
course_students = course.students

多对多关系的操作同样简洁，append 和列表操作就能管理关联关系。

---

六、常见问题与解决方案

6.1 JOIN 查询后 JSON 序列化失败

初学者经常遇到的一个问题是：使用 db.query(ModelA, ModelB).join(...).all() 后，返回的是一堆元组，FastAPI 无法正确序列化为 JSON。

错误示例：

# ❌ 这会返回元组列表，无法直接序列化
results = db.query(Post, func.count(Vote.post_id)).join(Vote).group_by(Post.id).all()
return results  # 报错：TypeError: cannot convert...

正确做法：

# ✅ 使用 .mappings() 获取字典格式结果
stmt = select(
    Post,
    func.count(Vote.post_id).label("votes_count")
).join(Vote, isouter=True).group_by(Post.id)

results = db.execute(stmt).mappings().all()
return results  # 返回的是字典列表，可正常序列化

mappings().all() 会将每行结果转换为字典格式，完全兼容 FastAPI 的响应序列化机制。

6.2 N+1 查询问题

当你查询一篇文章列表，然后遍历每篇文章去获取它的作者时，可能会产生 N+1 次数据库查询：

# ❌ 可能产生 N+1 次查询
posts = db.query(Post).all()
for post in posts:
    print(post.author.username)  # 每循环一次，就查一次数据库

解决方案：使用 joinedload 预加载关联数据：

from sqlalchemy.orm import joinedload

# ✅ 一次性加载所有关联数据
stmt = select(Post).options(joinedload(Post.author))
posts = db.execute(stmt).scalars().unique().all()

joinedload 会在执行查询时通过 SQL JOIN 一次性把关联的作者数据也加载进来，避免后续的重复查询。

---

七、完整项目结构推荐

在实际项目中，建议按照以下结构组织代码：

myapp/
├── models.py          # SQLAlchemy/SQLModel 模型定义
├── schemas.py         # Pydantic 请求/响应模型
├── database.py        # 数据库引擎和会话配置
├── crud.py            # 数据库操作函数
├── routers/
│   ├── users.py       # 用户相关路由
│   └── posts.py       # 文章相关路由
└── main.py            # FastAPI 应用入口

这样的分离让代码职责清晰，便于维护和测试。

---

八、总结

数据库表间的连接操作是 FastAPI 应用开发的核心技能之一。回顾本文的核心要点：

知识点	关键内容
外键	用 ForeignKey 建立表与表之间的物理关联
一对多	在“多”方定义外键，用 relationship 和 back_populates 建立双向关系
多对多	创建中间表，用 secondary 参数定义关系
响应模型	用 Pydantic 的 from_attributes=True 支持 ORM 对象序列化
预加载	用 joinedload 解决 N+1 查询问题
序列化问题	用 .mappings().all() 处理 JOIN 查询的结果