当前位置：首页 > news >正文

【FastAPI高级实战】结合查询参数与SQLModel Joins实现高效多表查询（分页、过滤、计数）

news 来源：原创 2025/6/16 6:29:48

想象一下，你正在开发一个超酷的Web应用，比如一个博客平台或者一个在线商店。你的API不仅要能把数据（比如文章列表、商品信息）展示给用户，更要聪明到能理解用户的各种“小心思”：用户可能想看最新的文章、搜索特定关键词的商品、或者只想看当前页面的10条数据等等。

这篇文章就是要带你解锁FastAPI中使用SQLModel来实现这些高级查询功能的秘籍，主要关注两大神技：查询参数（Query Parameters） 和 数据库联接（Joins）。

1. 查询参数：让你的API听懂用户的“指令”

什么是查询参数？简单说，它就像是你给API发出的“指令”，通过URL告诉服务器你想要什么样的数据。

举个例子：

你想看第2页的文章，每页显示5篇：http://localhost:8000/posts?skip=5&limit=5
你想搜索标题包含“夏天”的文章：http://localhost:8000/posts?search=夏天

这里的 ?skip=5&limit=5 和 ?search=夏天 就是查询参数。它们以问号 (?) 开头，用 key=value 的形式表示，多个参数之间用与号 (&) 连接。

查询参数能干啥？

过滤 (Filtering)：就像筛子一样，根据特定条件筛选数据。例如，只看某个作者的文章。
排序 (Sorting)：按发布时间、价格等给结果排序（虽然本文不细讲，但原理类似）。
分页 (Pagination)：当数据太多时，一页一页地看。limit（每页几条）和 skip（跳过几条，即 (页码-1) * 每页条数）是分页的黄金搭档。

FastAPI + SQLModel 如何实现查询参数？

在FastAPI中，这简直太简单了！你只需要在你的API接口函数（路径操作函数）里定义参数，并给它们加上类型提示。FastAPI会自动把URL中的查询参数“抓”出来，并转换成你想要的类型。

# --- main.py (或者你的API路由文件，比如 posts_router.py) ---
from typing import Optional, List
from fastapi import APIRouter, Depends
from sqlmodel import Session, select, func # 导入SQLModel的核心组件
from .. import schemas, models, database, oauth2 # 假设你的项目结构router = APIRouter(prefix="/posts", # 给这个路由下的所有路径加上 /posts 前缀tags=["Posts"]   # 在API文档中分组显示
)# 假设你的 database.py 里面有 get_db 函数来获取数据库会话
# from .database import get_db@router.get("/", response_model=List[schemas.PostOut]) # 注意这里的 response_model
async def get_all_posts(db: Session = Depends(database.get_db), # 依赖注入，获取数据库会话limit: int = 10,  # 默认每页10条skip: int = 0,    # 默认从第0条开始（跳过0条）search: Optional[str] = None # 可选的搜索词，默认为None# current_user: models.User = Depends(oauth2.get_current_user) # 如果需要登录认证
):"""获取所有帖子，支持分页和搜索，并统计每个帖子的点赞数。"""# 基础查询语句，选择帖子模型和统计投票数statement = (select(models.Post, # 我们要查询帖子本身func.count(models.Vote.post_id).label("votes") # 还要统计每个帖子的投票数，并命名为 "votes")# 关键：使用 .join() 来连接 Post 表和 Vote 表# isouter=True 表示使用 LEFT OUTER JOIN，这样即使帖子没有投票，也会被查询出来，投票数为0.join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True)# 按帖子ID分组，这样 func.count 才能正确统计每个帖子的投票.group_by(models.Post.id))# 如果用户提供了搜索词，就在查询语句中加入过滤条件if search:# models.Post.title.contains(search) 表示标题包含搜索词即可# 对于大小写不敏感的搜索，可以使用 .ilike(f"%{search}%")，但要注意数据库是否支持statement = statement.where(models.Post.title.contains(search))# 应用分页参数statement = statement.limit(limit).offset(skip)# 执行查询# db.exec(statement).all() 会返回一个结果列表，# 列表中的每个元素是一个元组 (Post对象, votes数量)results = db.exec(statement).all()# FastAPI 会根据 response_model=List[schemas.PostOut] 自动处理这个 results# schemas.PostOut 需要被定义成能接收 Post 对象和 votes 数量的结构return results

代码讲解：

@router.get("/", response_model=List[schemas.PostOut])：
- 定义了一个GET请求的接口，路径是 /posts/ (因为router有prefix="/posts")。
- response_model=List[schemas.PostOut] 告诉FastAPI，这个接口返回的数据会是一个列表，列表里每个元素的结构都符合 schemas.PostOut 这个我们接下来要定义的“响应模型”。
async def get_all_posts(...)：
- db: Session = Depends(database.get_db): FastAPI的依赖注入。get_db函数（通常在database.py中定义）会提供一个数据库会话 db，让我们能和数据库打交道。
- limit: int = 10, skip: int = 0, search: Optional[str] = None: 这些就是我们的查询参数！FastAPI会从URL中解析它们。比如用户访问 /posts?limit=5&search=python，那么 limit就是5，search就是"python"。
statement = select(models.Post, func.count(models.Vote.post_id).label("votes")) ...:
- 这里开始构建SQLModel的查询语句。select() 是起点。
- 我们不仅要 models.Post（帖子本身），还要 func.count(models.Vote.post_id).label("votes")（统计每个帖子的投票数，并把这个统计结果命名为 votes）。func 来自 sqlalchemy (SQLModel底层使用它)，提供了数据库函数如 COUNT, SUM等。
.join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True):
- 这是多表查询的关键！我们要把 Post 表和 Vote 表连接起来。
- 连接条件是 models.Vote.post_id == models.Post.id (投票表中的帖子ID等于帖子表中的帖子ID)。
- isouter=True 表示使用 左外连接 (LEFT OUTER JOIN)。这意味着即使一个帖子没有任何投票记录，它仍然会出现在结果中，其 votes 计数会是0。如果不用 isouter=True (即默认的内连接 INNER JOIN)，那么没有投票的帖子就不会被查出来。
.group_by(models.Post.id):
- 因为我们用了聚合函数 func.count()，所以需要告诉数据库按什么来分组统计。这里我们按帖子的ID (models.Post.id) 分组，这样就能得到每个帖子的投票数。
if search: statement = statement.where(models.Post.title.contains(search)):
- 如果用户在URL中提供了 search 参数，我们就在查询语句中添加一个 where 条件，筛选出标题 (models.Post.title) 包含 (contains) 搜索词的帖子。
statement = statement.limit(limit).offset(skip):
- 将分页参数应用到查询语句上。
results = db.exec(statement).all():
- 执行最终构建好的查询语句，并获取所有结果。results 会是一个列表，每个元素是 (Post对象, votes数量) 这样的元组。
return results:
- 直接返回这个 results。FastAPI会很智能地根据我们之前定义的 response_model=List[schemas.PostOut] 来把这个元组列表转换成符合 PostOut 结构的JSON列表返回给客户端。

2. 数据库表关系与联接：让数据“手拉手”

在真实世界中，数据很少是孤零零存在的。比如：

一个用户（User） 可以发布多篇帖子（Post）。（一对多关系）
一个用户（User） 可以给多篇帖子（Post） 点赞，一篇帖子（Post） 也可以被多个用户（User） 点赞。（多对多关系，通常通过一个中间表，如 投票（Vote） 表来实现）

SQLModel 如何定义这些关系？

SQLModel 的美妙之处在于它同时是Pydantic模型（用于数据校验和序列化）和数据库表模型。

# --- models.py ---
from datetime import datetime
from typing import Optional, List # 注意这里也需要 List
from sqlmodel import SQLModel, Field, Relationship # 导入SQLModel的核心组件
from sqlalchemy import text # 用于设置数据库级别的默认值# 用户模型
class User(SQLModel, table=True):__tablename__ = "users" # 表名id: Optional[int] = Field(default=None, primary_key=True)email: str = Field(unique=True, index=True, nullable=False) # 邮箱唯一且建立索引password: str = Field(nullable=False)created_at: datetime = Field(default_factory=datetime.utcnow, # Python级别默认值sa_column_kwargs={"server_default": text("now()")} # 数据库级别默认值)# 定义关系：一个用户可以有多篇帖子# "Post" 是关联的模型类名（字符串形式避免循环导入问题）# back_populates="owner" 指向 Post 模型中名为 "owner" 的关系属性posts: List["Post"] = Relationship(back_populates="owner")# 帖子模型
class Post(SQLModel, table=True):__tablename__ = "posts" # 表名id: Optional[int] = Field(default=None, primary_key=True)title: str = Field(index=True, nullable=False) # 标题也加个索引，方便搜索content: str = Field(nullable=False)published: bool = Field(default=True, sa_column_kwargs={"server_default": text("true")})created_at: datetime = Field(default_factory=datetime.utcnow,sa_column_kwargs={"server_default": text("now()")})# 外键：这篇帖子的作者是谁# foreign_key="users.id" 指向 users 表的 id 字段# nullable=False 表示每篇帖子都必须有作者owner_id: int = Field(foreign_key="users.id", nullable=False)# 定义关系：这篇帖子的作者 (User 对象)# back_populates="posts" 指向 User 模型中名为 "posts" 的关系属性owner: Optional[User] = Relationship(back_populates="posts")# 投票模型 (用于用户给帖子点赞)
class Vote(SQLModel, table=True):__tablename__ = "votes"# 复合主键：一个用户对一个帖子只能投一票user_id: int = Field(foreign_key="users.id", primary_key=True, ondelete="CASCADE")post_id: int = Field(foreign_key="posts.id", primary_key=True, ondelete="CASCADE")# ondelete="CASCADE" 表示如果关联的 User 或 Post 被删除，这条 Vote 记录也会被自动删除

代码讲解 (models.py):

class User(SQLModel, table=True):: 定义一个User模型，它既是SQLModel（数据库表模型），也是Pydantic模型。table=True表示它对应数据库中的一张表。
id: Optional[int] = Field(default=None, primary_key=True): 定义 id 字段，是主键，可选（数据库会自动生成）。
posts: List["Post"] = Relationship(back_populates="owner"):
- 这是定义关系的关键！它告诉SQLModel，一个 User 对象可以关联多个 Post 对象。
- List["Post"] 表示这个 posts 属性是一个 Post 对象的列表。用字符串 "Post" 是为了避免Python在解析时可能遇到的循环导入问题。
- back_populates="owner" 指的是，在 Post 模型那边，有一个名为 owner 的属性也定义了与 User 的关系，并且它们是相互关联的。
class Post(SQLModel, table=True):: 类似地定义 Post 模型。
owner_id: int = Field(foreign_key="users.id", nullable=False):
- 这是外键字段。它存储的是对应 User 表中某条记录的 id。
- foreign_key="users.id" 明确指定了它引用 users 表的 id 列。
- nullable=False 表示一篇帖子必须有一个作者。
owner: Optional[User] = Relationship(back_populates="posts"):
- 与 User 模型中的 posts 关系相呼应。它表示一个 Post 对象关联一个 User 对象（即帖子的作者）。
class Vote(SQLModel, table=True)::
- user_id: int = Field(foreign_key="users.id", primary_key=True, ...)
- post_id: int = Field(foreign_key="posts.id", primary_key=True, ...)
- 这两个字段共同构成了复合主键，确保一个用户对一篇帖子只能投票一次。它们也都是外键，分别引用 users 表和 posts 表。
- ondelete="CASCADE": 这是一个数据库层面的约束。如果一个用户被删除了，那么他所有的投票记录也会自动被删除。同理，如果一个帖子被删除了，关于这个帖子的所有投票记录也会被删除。这有助于保持数据的整洁和一致性。

为何需要联接 (Joins)？

想象一下，你想显示一篇帖子的详细信息，同时还要显示发帖人的用户名，以及这篇帖子有多少个赞。

帖子标题、内容在 Post 表。
发帖人用户名在 User 表（通过 Post.owner_id 关联）。
点赞数需要统计 Vote 表中对应 post_id 的记录数量。

如果不用联接，你可能需要：

查询 Post 表获取帖子信息。
根据 Post.owner_id 再去 User 表查询用户信息。
根据 Post.id 再去 Vote 表统计点赞数。

这样查询次数太多，效率低下！联接 (Join) 就是为了解决这个问题，它允许你在一次数据库查询中，把来自不同但相关联的表的数据“拼接”在一起。

3. 玩转 SQLModel：多表联合查询实战

我们回到之前的 get_all_posts 接口，它已经用到了联接：

# --- main.py (或者你的API路由文件) ---
# ... (省略之前的导入和router定义) ...# 先定义好我们的响应模型 schemas.py
# --- schemas.py ---
from datetime import datetime
from pydantic import BaseModel # SQLModel本身就是Pydantic模型，但有时为了清晰或特定场景会用BaseModel
from .models import User # 导入你的SQLModel模型 Userclass UserOut(BaseModel): # 用于在响应中展示的用户信息，不包含密码等敏感信息id: intemail: strcreated_at: datetimeclass Config:from_attributes = True # Pydantic V2中的配置，允许从ORM对象属性创建模型实例class PostResponseBase(BaseModel): # 帖子基础信息id: inttitle: strcontent: strpublished: boolcreated_at: datetimeowner_id: intowner: UserOut # 嵌套UserOut，显示作者信息class Config:from_attributes = Trueclass PostOut(BaseModel): # 最终API返回的帖子结构，包含帖子信息和投票数Post: PostResponseBase # 帖子自身的详细信息votes: int             # 这个帖子的投票总数class Config:from_attributes = True # 确保可以从 (Post对象, votes数量) 这样的元组构造# ... 回到你的 API 路由文件 ...
@router.get("/", response_model=List[schemas.PostOut])
async def get_all_posts(db: Session = Depends(database.get_db),limit: int = 10,skip: int = 0,search: Optional[str] = None
):statement = (select(models.Post, # 选择 Post 模型对象func.count(models.Vote.post_id).label("votes") # 计算投票数，并命名为 "votes").join(models.Vote, models.Post.id == models.Vote.post_id, isouter=True) # 左外连接 Vote 表.group_by(models.Post.id) # 按帖子ID分组)if search:statement = statement.where(models.Post.title.contains(search))statement = statement.limit(limit).offset(skip)# results_from_db 的结构是 List[Tuple[Post, int]]# 例如：[(<Post object at 0x...>, 5), (<Post object at 0x...>, 0)]results_from_db = db.exec(statement).all()# FastAPI 会自动根据 response_model 将 results_from_db 转换# 它会尝试把每个 (Post, int) 元组 构造成一个 schemas.PostOut 对象# Post对象会被用来填充 PostOut.Post 字段 (内部的 PostResponseBase 会通过 Post.owner 自动加载用户信息)# int 会被用来填充 PostOut.votes 字段return results_from_db# 如果你想创建一个帖子，同时关联当前登录的用户作为作者
@router.post("/", status_code=status.HTTP_201_CREATED, response_model=schemas.PostResponseBase) # 注意响应模型
def create_post(post_data: schemas.PostCreate, # PostCreate 是一个只包含 title, content, published 的Pydantic模型db: Session = Depends(database.get_db),current_user: models.User = Depends(oauth2.get_current_user) # 获取当前登录用户
):# **post_data.model_dump() 把 Pydantic 模型转为字典# owner_id=current_user.id 把当前用户的ID设为帖子的作者IDnew_post = models.Post(**post_data.model_dump(), owner_id=current_user.id)db.add(new_post)db.commit()db.refresh(new_post) # 刷新new_post对象，使其包含数据库生成的值（如ID, created_at）# new_post.owner 会自动通过 relationship 加载关联的 User 对象# FastAPI 会根据 response_model=schemas.PostResponseBase 自动序列化return new_post

代码讲解 (schemas.py 和 create_post):

schemas.py 的作用：
- UserOut: 定义了当我们需要在API响应中显示用户信息时，只显示哪些字段（比如不显示密码）。
- PostResponseBase: 定义了帖子的基本输出信息，并且它嵌套了 owner: UserOut，这意味着在返回帖子信息时，会自动把作者的 UserOut 信息也包含进去。
- PostOut: 这是我们 get_all_posts 接口最终的响应结构。它包含一个 Post 字段（类型是 PostResponseBase，即帖子详情加作者信息）和一个 votes 字段（帖子的点赞数）。
- class Config: from_attributes = True: 这个配置（在Pydantic V2中，旧版是 orm_mode = True）非常重要。它允许Pydantic模型直接从数据库对象（ORM对象，比如我们的SQLModel实例）的属性来创建实例。比如，如果一个 Post SQLModel对象有 title 和 content 属性，PostResponseBase 就能直接用这些属性来填充自己。
get_all_posts 返回 results_from_db：
- 当 db.exec(statement).all() 执行后，results_from_db 是一个列表，每个元素是 (Post模型实例, 投票数) 这样的元组。
- FastAPI看到 response_model=List[schemas.PostOut]，它会尝试把每个元组 (p, v) 转换成一个 schemas.PostOut(Post=p, votes=v) 的实例。
- schemas.PostOut 的 Post 字段类型是 schemas.PostResponseBase。由于 from_attributes = True，PostResponseBase 会从 p (Post模型实例) 中读取属性。
- 特别地，PostResponseBase 中的 owner: UserOut 字段，会因为 Post SQLModel模型中定义了 owner: Optional[User] = Relationship(...) 而被自动填充。SQLModel（或底层的SQLAlchemy）会在需要时加载关联的 User 对象，然后 UserOut 再从这个 User 对象中提取信息。
create_post 接口：
- post_data: schemas.PostCreate: PostCreate 是一个Pydantic模型，用于接收创建帖子时客户端发来的数据（比如只有 title, content）。
- current_user: models.User = Depends(oauth2.get_current_user): 假设你有一个 oauth2.py 文件，里面的 get_current_user 函数会验证JWT token并返回当前登录的 User SQLModel对象。
- new_post = models.Post(**post_data.model_dump(), owner_id=current_user.id):
  - post_data.model_dump() 将 PostCreate Pydantic模型转换成一个字典。
  - ** 将这个字典解包，作为参数传递给 models.Post 的构造函数。
  - owner_id=current_user.id 明确设置了这篇新帖子的作者是当前登录用户。
- db.add(new_post), db.commit(), db.refresh(new_post): 这是标准的SQLModel（或SQLAlchemy）操作，将新对象添加到会话、提交到数据库、然后刷新对象以获取数据库生成的值。
- return new_post: 返回创建好的 models.Post 对象。因为 response_model=schemas.PostResponseBase，FastAPI会自动将这个 Post 对象（包括其通过relationship加载的 owner 信息）转换为 PostResponseBase 格式的JSON响应。

关键点总结：

SQLModel 让模型定义更简单：一个类同时搞定Pydantic校验和数据库表结构。
Relationship 很强大：在SQLModel模型中定义好 Relationship，SQLModel（底层SQLAlchemy）就能帮你处理很多关联数据的加载。
select().join() 是多表查询的核心：用它来连接不同的表。isouter=True 用于左外连接，确保即使没有关联数据（如帖子没有投票）主表数据也能查出来。
func.count() 和 group_by() 用于聚合统计：比如统计每个帖子的投票数。
FastAPI 的 response_model 和 Pydantic 的 from_attributes = True 是天作之合：它们能让你轻松地将数据库查询结果（甚至是包含关联对象的复杂结果）转换成规范的JSON API响应。

总结

通过本文，你已经掌握了如何在FastAPI中利用SQLModel的强大功能，通过查询参数让API更灵活，通过数据库联接和Relationship高效查询和展示关联数据。

你学会了：

在FastAPI接口中定义查询参数（limit, skip, search）。
使用SQLModel的 select() 语句，并通过 .where(), .limit(), .offset() 来应用这些参数。
在SQLModel模型中通过 Field(foreign_key=...) 和 Relationship(back_populates=...) 定义表间关系。
使用 select().join() 进行多表查询，特别是用 isouter=True 实现左外连接。
使用 func.count().label() 和 .group_by() 进行聚合统计。
设计合适的Pydantic响应模型（如 schemas.PostOut），并利用FastAPI的 response_model 特性自动转换查询结果。