技术文档

Appearance

21. FastAPI 使用 SQLAlchemy 2 Async 映射类定义关系（一对一、一对多、多对多） ​

一套模型理解三种关系 ​

  • 一对一：一个用户最多有一份资料，资料表的 user_id 必须唯一。
  • 一对多：一个用户有多篇文章，外键放在“多”的一方。
  • 多对多：一个用户有多个角色，一个角色也能给多个用户，通过关联表连接。

数据库外键保证引用有效；relationship 提供 Python 对象之间的导航。仅声明 relationship 不会替你创建所有外键和唯一约束。

完整的 models.py ​

沿用第 20 篇的 database.py，用以下完整内容替换 models.py。后续三篇都导入这些模型，不要重复定义同名表。第 20 篇的 User 字段与这里一致，本篇继续补充三类关系。

如果此前按旧版教程建过结构不同的 users 表，create_all() 不会自动增加或修改已有列；请使用 Alembic 做迁移，或只在全新的练习数据库中重建表，不要直接删除含有真实数据的表。

python
from __future__ import annotations
from sqlalchemy import Boolean, Column, ForeignKey, String, Table, Text, true
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

user_roles = Table(
    "user_roles", Base.metadata,
    Column("user_id", ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
    Column("role_id", ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True),
    mysql_engine="InnoDB", mysql_charset="utf8mb4",
)

class User(Base):
    __tablename__ = "users"
    __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(254), unique=True)
    active: Mapped[bool] = mapped_column(Boolean, default=True, server_default=true())
    batch_tag: Mapped[str | None] = mapped_column(String(32), index=True)
    profile: Mapped[Profile | None] = relationship(
        back_populates="user", cascade="all, delete-orphan",
        single_parent=True, passive_deletes=True,
    )
    posts: Mapped[list[Post]] = relationship(
        back_populates="user", cascade="all, delete-orphan", passive_deletes=True,
    )
    roles: Mapped[list[Role]] = relationship(
        secondary=user_roles, back_populates="users", passive_deletes=True,
    )

class Profile(Base):
    __tablename__ = "profiles"
    __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), unique=True,
    )
    bio: Mapped[str] = mapped_column(String(200), default="")
    user: Mapped[User] = relationship(back_populates="profile")

class Post(Base):
    __tablename__ = "posts"
    __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), index=True,
    )
    title: Mapped[str] = mapped_column(String(100))
    content: Mapped[str] = mapped_column(Text, default="")
    user: Mapped[User] = relationship(back_populates="posts")

class Role(Base):
    __tablename__ = "roles"
    __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(64), unique=True)
    users: Mapped[list[User]] = relationship(
        secondary=user_roles, back_populates="roles", passive_deletes=True,
    )

创建 init_db.py。异步引擎不能直接调用同步的 Base.metadata.create_all(engine)，需要通过异步连接的 run_sync() 执行：

python
import asyncio

from database import master_engine
from models import Base


async def init_db() -> None:
    async with master_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    # 独立脚本结束前释放连接池
    await master_engine.dispose()
    print("学习表创建完成")


if __name__ == "__main__":
    asyncio.run(init_db())

执行 uv run python init_db.py。确保导入模型后再调用 create_all，否则元数据里可能没有对应表。asyncio.run() 只放在脚本入口，不要在已经运行事件循环的 FastAPI 路由中再次调用。

看懂核心写法 ​

Mapped[int] 声明 Python 属性；mapped_column 配置数据库列。Mapped[str | None] 通常对应允许空值的列。Python 默认值 default 与数据库默认值 server_default 作用位置不同；通过其他程序直接插入时，只有数据库默认值仍会生效。

back_populates 将两端关系对应起来，便于对象状态同步。标量关系表达一对一，列表关系表达集合；一对一最终仍要靠数据库 unique=True 阻止两条资料指向同一用户。

关联表的联合主键保证同一用户和角色不会重复关联。若关联还需要“授予时间”“授予人”等字段，适合把关联表映射成独立的关联对象类，而不是仅使用 secondary。

异步环境不能隐式延迟加载关系 ​

relationship 的声明方式与同步 SQLAlchemy 相同，但异步 Session 不允许在普通属性访问时偷偷发起数据库 I/O。先查询用户、再在异步函数中直接访问一个尚未加载的 user.posts，可能抛出 MissingGreenlet，而不是像同步代码那样自动补发查询。

查询时应使用 joinedload、selectinload 明确预加载：

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

from database import SessionLocal
from models import User


async def load_user(user_id: int) -> None:
    async with SessionLocal() as db:
        user = await db.scalar(
            select(User)
            .options(
                joinedload(User.profile),
                selectinload(User.posts),
                selectinload(User.roles),
            )
            .where(User.id == user_id)
        )

        if user is not None:
            # 三组关系已经加载，不会再触发隐式数据库查询
            print(user.profile, len(user.posts), len(user.roles))

也可以在已有对象上使用 await db.refresh(user, attribute_names=["posts"]) 显式加载指定关系。第 23、24 篇会分别使用这两种思路；不要把同步 ORM 的隐式 lazy load 习惯直接搬进异步代码。

级联必须区分两层 ​

ORM 的 cascade="all, delete-orphan" 处理对象操作；数据库的 ON DELETE CASCADE 处理数据库行的引用。批量 DELETE 不会逐个执行 ORM 对象级联，所以两层规则要协调。passive_deletes=True 表示可以让数据库处理部分关联清理，不是独立的数据库删除规则。

用户删除时，资料、文章和关联行跟随删除；共享角色本身保留。不要在多对多双方随意启用删除实体的级联，否则可能沿关系扩散删除整张对象图。

更换一对一资料时，直接新建另一份可能遇到唯一键写入顺序问题。通常更新现有资料；确需替换时，在同一事务内明确删除旧资料并 await db.flush()，再添加新资料。

练习：查看表结构，找出外键、唯一键、索引和联合主键；尝试为同一用户插入两份资料，确认数据库约束确实生效。

网站备案号:津ICP备15001525号-2