数据库进阶 / 已完成

SQLAlchemy 关系查询和性能优化

理解一对多、多对多、relationship、N+1 查询、selectinload、joinedload 和分页性能。

返回文章积累

一句话:SQLAlchemy 关系查询就是让表和表之间能互相找到对方,性能优化就是避免一次页面请求背后偷偷查很多次数据库。

本篇学完你会什么:理解一对多、多对多、外键、relationship、懒加载、selectinload、joinedload、N+1 查询和分页查询这些真实项目里最常见的数据库进阶问题。

1. 为什么 CRUD 之后要学关系查询

CRUD 文章里我们主要操作一张表:

users 用户表

但真实项目里,很少只有一张表。

比如用户管理系统会有:

users 用户表
roles 角色表
permissions 权限表
articles 文章表
comments 评论表

这时你要回答的问题会变成:

这个用户有哪些角色?
这个角色有哪些权限?
这篇文章是谁写的?
这篇文章有多少评论?
查询列表时要不要一起查作者?

这些就是关系查询。

2. 一对多是什么

一对多就是:

一个用户可以写多篇文章
一篇文章只属于一个作者

数据库里通常这样设计:

users
- id
- username

articles
- id
- title
- author_id

articles.author_id 指向 users.id,这就是外键。

SQLAlchemy 模型可以这样写:

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(32), unique=True)

    articles: Mapped[list["Article"]] = relationship(back_populates="author")


class Article(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

    author: Mapped["User"] = relationship(back_populates="articles")

这里先只展示关键 import。Base 一般来自前面文章里的数据库配置,例如 DeclarativeBase;实际项目里会放在统一的 db/base.pycore/database.py

大白话:

字段意思
author_id数据库里真正保存的关联 id
author从文章找到作者
articles从用户找到他的文章
back_populates两边互相说明对方是谁

3. 多对多是什么

多对多就是:

一个用户可以有多个角色
一个角色也可以分给多个用户

数据库通常加一张中间表:

user_roles
- user_id
- role_id

模型示例:

from sqlalchemy import Column, ForeignKey, String, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship


user_roles = Table(
    "user_roles",
    Base.metadata,
    Column("user_id", ForeignKey("users.id"), primary_key=True),
    Column("role_id", ForeignKey("roles.id"), primary_key=True),
)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(32))
    roles: Mapped[list["Role"]] = relationship(secondary=user_roles, back_populates="users")


class Role(Base):
    __tablename__ = "roles"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32))
    users: Mapped[list["User"]] = relationship(secondary=user_roles, back_populates="roles")

中间表就像登记表:谁拿了哪张角色卡,都记在这里。

4. SQLAlchemy 里的 relationship 是什么

relationship 不是数据库字段。

真正存在数据库里的通常是:

author_id
user_id
role_id

relationship 是 SQLAlchemy 帮你在 Python 里更方便地访问关系:

article.author.username
user.roles
role.users

所以要记住:

东西在数据库里吗作用
ForeignKey保存真实关联
relationship不是直接字段让 Python 代码更容易访问关联对象

5. 查询用户和文章怎么写

查询文章列表:

stmt = select(Article).order_by(Article.id.desc())
result = await db.execute(stmt)
articles = result.scalars().all()

如果页面还要显示作者名,你可能会写:

for article in articles:
    print(article.author.username)

这看起来很自然,但可能埋下性能问题。

在同步 SQLAlchemy 里,这种写法可能偷偷触发懒加载:循环一次,就多查一次作者。

在异步 SQLAlchemy 里,问题还不只是慢。因为异步代码不能随便在属性访问时偷偷发起数据库 IO,直接访问未提前加载的 relationship 还可能遇到 MissingGreenlet 或类似的隐式 IO 错误。

所以在 FastAPI 这类异步接口里,更推荐在 select() 阶段就明确告诉 SQLAlchemy:这次要提前加载哪些关系。

6. N+1 查询是什么

N+1 查询就是:

先查 1 次文章列表
再为 N 篇文章分别查作者

假设列表有 20 篇文章:

1 次查询文章
20 次查询作者
= 21 次查询

这就是 N+1。

页面上只是一个列表,但数据库背后可能被查了很多次。数据少时没感觉,数据多了就会慢。

7. selectinload 和 joinedload 怎么选

解决 N+1,常用 eager loading,也就是提前把关联数据查好。

下面示例会用到这些查询相关的 import:

from sqlalchemy import select, func
from sqlalchemy.orm import joinedload, selectinload

selectinload

stmt = select(Article).options(selectinload(Article.author))
result = await db.execute(stmt)
articles = result.scalars().all()

大白话:

先查文章
再用一条额外查询把这些文章的作者一起查出来

适合列表页,很常用。

joinedload

stmt = select(Article).options(joinedload(Article.author))
result = await db.execute(stmt)
articles = result.scalars().all()

大白话:

用 JOIN 一次把文章和作者查出来

适合一对一、多对一这种不会让结果行膨胀太夸张的场景。

选择建议:

场景推荐
列表页查多条记录和关联对象selectinload
详情页查一个对象和少量关联joinedloadselectinload
多对多、关联很多优先 selectinload
只要关联 id,不要对象详情不一定需要 relationship 加载

8. 分页和排序怎么避免变慢

分页查询常见写法:

stmt = (
    select(Article)
    .where(Article.is_published == True)
    .order_by(Article.id.desc())
    .offset((page - 1) * page_size)
    .limit(page_size)
)

要注意三件事:

  1. 排序字段最好有索引。
  2. 不要一次查太多字段和太多关联。
  3. total 计数可以单独查,不要把整个列表拿回来再 len()

计数:

count_stmt = select(func.count()).select_from(Article).where(Article.is_published == True)
total = await db.scalar(count_stmt)

大白话:列表页要像仓库取货,一页拿一箱,不要为了看第一页就把整个仓库搬出来。

9. 常见错误

错误后果修正
只写 relationship,不写 ForeignKey数据库不知道怎么关联外键是真关联
列表页循环访问关联对象N+1 查询selectinload
异步项目里直接访问未加载的 relationship可能 MissingGreenlet 或隐式查询查询时提前 selectinload / joinedload
多对多不用中间表关系表达不清楚user_roles 这类关系表
分页后又在 Python 里过滤数据越多越慢过滤条件放 SQL
一次加载太多关联返回很慢页面需要什么就查什么
忘记索引查询越来越慢给查询条件和排序字段加合适索引

10. 检查清单

[ ] 一对多是否有清楚的外键字段
[ ] 多对多是否有中间表
[ ] relationship 两边是否 back_populates 对齐
[ ] 列表页是否避免 N+1 查询
[ ] 是否知道什么时候用 selectinload
[ ] 分页查询是否在数据库层完成
[ ] 排序和过滤字段是否考虑索引
[ ] 返回给前端的数据是否只包含页面需要的字段

11. 总结表

名词大白话
外键一张表指向另一张表的 id
一对多一个用户多篇文章
多对多一个用户多个角色,一个角色多个用户
中间表记录多对多关系的登记表
relationshipPython 里方便访问关联对象的工具
N+1查一次列表,又为每条记录额外查一次
selectinload适合列表页的提前加载
joinedload用 JOIN 一起查关联对象
分页一次只取一页数据

上一篇建议:大白话讲解——后台权限系统设计.md

下一篇建议:大白话讲解——FastAPI 项目配置、多环境和部署上线.md