36｜SQLAlchemy 模型关系与查询优化

真实项目中的表不是孤立的：服务器属于某个项目，项目有多个任务，任务关联到具体服务器。SQLAlchemy 用关系（relationship）表达这些关联，用外键（ForeignKey）在数据库层面维护引用完整性。同时，查询方式的选择会直接影响接口性能——这篇把关系和查询优化一起讲。

一、外键与关系

一对多

一个项目有多台服务器：

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

class Project(Base):
    __tablename__ = "projects"

    id = Column(Integer, primary_key=True)
    name = Column(String(64), nullable=False)

    # 关系定义：一个 Project 关联多个 Server
    servers = relationship("Server", back_populates="project")

class Server(Base):
    __tablename__ = "servers"

    id = Column(Integer, primary_key=True)
    hostname = Column(String(64), nullable=False)
    project_id = Column(Integer, ForeignKey("projects.id"))

    # 反向关系：一台 Server 属于一个 Project
    project = relationship("Project", back_populates="servers")

概念	定义位置	作用
ForeignKey	多的一端（Server）	数据库外键约束，维护引用完整性
relationship	两端都定义	ORM 层面的对象关联，不创建数据库列
back_populates	两端互相指向	双向导航：project.servers 和 server.project

使用：

project = db.get(Project, 1)
for server in project.servers:
    print(server.hostname)

多对多

服务器可以有多个标签，标签也可以对应多台服务器：

server_tags = Table(
    "server_tags",
    Base.metadata,
    Column("server_id", Integer, ForeignKey("servers.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tags"
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False, unique=True)

class Server(Base):
    __tablename__ = "servers"
    id = Column(Integer, primary_key=True)
    hostname = Column(String(64), nullable=False)

    tags = relationship("Tag", secondary=server_tags, back_populates="servers")

多对多关系需要一张中间表（关联表），只存两个外键，没有自己的主键（复合主键由两个外键组成）。

二、查询方式

懒加载（Lazy Loading）

默认方式，访问关系属性时才发 SQL：

project = db.get(Project, 1)     # 执行 1 条 SQL
for s in project.servers:        # 执行 N 条 SQL（每台服务器一条）
    print(s.hostname)

这就是经典的 "N+1 查询" 问题：1 条查项目，N 条查关联的服务器。数据量大时性能极差。

急加载（Eager Loading）

用 selectinload 一次性把关联数据加载出来：

from sqlalchemy.orm import selectinload

stmt = select(Project).options(selectinload(Project.servers))
project = db.execute(stmt).scalar_one()

for s in project.servers:   # 不再发额外 SQL
    print(s.hostname)

执行 2 条 SQL：

1. SELECT * FROM projects WHERE id = 1
2. SELECT * FROM servers WHERE project_id IN (1)

加载方式	用法	适用场景
selectinload	对多的一端，子查询加载	一对多、多对多
joinedload	JOIN 加载	对一的一端
subqueryload	子查询加载	复杂嵌套关系

直接 JOIN 查询

不需要加载完整对象，只取需要的字段：

from sqlalchemy import select

stmt = (
    select(Server.hostname, Project.name)
    .join(Project, Server.project_id == Project.id)
    .where(Server.status == "running")
)

results = db.execute(stmt).all()
for hostname, project_name in results:
    print(f"{hostname} ({project_name})")

三、查询优化建议

只查询需要的列

# 不推荐：SELECT *，加载所有列
stmt = select(Server)

# 推荐：只选需要的列
stmt = select(Server.id, Server.hostname)

限制返回数量

stmt = select(Server).limit(20).offset(0)

用 count 替代 len()

from sqlalchemy import func

# 错误：先把所有数据加载到内存再计数
count = len(db.execute(select(Server)).scalars().all())

# 正确：数据库层面计数
stmt = select(func.count()).select_from(Server)
count = db.execute(stmt).scalar()

批量操作

# 错误：循环插入，每次一条 INSERT
for s in servers:
    db.add(s)
db.commit()

# 正确：批量插入
from sqlalchemy import insert

db.execute(insert(Server), [
    {"hostname": "web-01", "ip": "192.168.1.10"},
    {"hostname": "web-02", "ip": "192.168.1.11"},
])
db.commit()

四、级联操作

删除项目时，自动删除关联的服务器：

class Project(Base):
    __tablename__ = "projects"
    id = Column(Integer, primary_key=True)
    name = Column(String(64))

    servers = relationship(
        "Server",
        back_populates="project",
        cascade="all, delete-orphan",
    )

级联值	作用
all	包含以下所有
delete	删除父对象时删除子对象
delete-orphan	子对象解除关联后删除
save-update	保存/更新父对象时级联到子对象

级联操作方便但危险。删除项目前务必确认业务上是否需要同时删除服务器。如果不确定，不要设置 cascade="delete"，改用手动删除或软删除（设置 deleted_at 字段）。

五、常见错误

循环导入

# project.py
from server import Server   # Server 里又 import Project

class Project(Base):
    servers = relationship("Server", ...)

解决方案：用字符串形式引用模型名，而不是直接导入类。

class Project(Base):
    servers = relationship("Server", back_populates="project")   # 字符串引用

N+1 查询

# 错误：懒加载导致 N+1
projects = db.execute(select(Project)).scalars().all()
for p in projects:
    for s in p.servers:   # 每个 project 都发一条 SQL 查 servers
        ...

# 正确：急加载
stmt = select(Project).options(selectinload(Project.servers))
projects = db.execute(stmt).scalars().all()

忘记设置外键

# 错误：只定义 relationship，没有 ForeignKey
class Server(Base):
    project = relationship("Project")   # 数据库层面没有外键约束

# 正确：relationship + ForeignKey 都要有
class Server(Base):
    project_id = Column(Integer, ForeignKey("projects.id"))
    project = relationship("Project", back_populates="servers")