FastAPI + SQLAlchemy ORM 表关系

模型关系分为：一对一、一对多、多对多。核心依赖：ForeignKey（数据库约束） + relationship（ORM 对象导航）。

一、一对一关系（One-to-One）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


from __future__ import annotations
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import Integer, String, ForeignKey
from db.base import Base

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(255))
    password: Mapped[str] = mapped_column(String(255))
    user_extension: Mapped["User_Extension"] = relationship(back_populates="user", uselist=False)


class User_Extension(Base):
    __tablename__ = "user_extension"  # 注意：原文误写为 __table__，应为 __tablename__

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    phone_number: Mapped[int] = mapped_column(Integer)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"), unique=True)  # 一对一必须加 unique=True
    user: Mapped["User"] = relationship(back_populates="user_extension")

关键说明

ForeignKey 的作用

在数据库层面创建外键约束，将 user_extension.user_id 关联到 user.id。
保证参照完整性：插入或更新时，值必须存在于 user.id。
SQLAlchemy 利用此外键配合 relationship 实现对象导航。

一对一必须在外键列加 unique=True，否则无法保证“一个用户最多一个扩展”。

relationship 的作用

在 Python 对象层面建立关联，不对应数据库列。
uselist=False：表示返回单个对象（非列表），是区分一对一与一对多的关键。
back_populates：实现双向关系。两端字段名必须互指，例如：
- User.user_extension ↔ User_Extension.user
- 设置任一端，另一端自动同步。

使用示例

1
2
3
4
5
6
7
8


user = User(name="Alice", password="123")
ext = User_Extension(phone_number=123456789, user=user)

# 或
user.user_extension = ext

print(user.user_extension.phone_number)  # 123456789
print(ext.user.name)                     # Alice

默认懒加载：访问 user.user_extension 时才触发 SQL 查询。

二、一对多关系（One-to-Many）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


# user.py
from __future__ import annotations
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import Integer, String, ForeignKey
from typing import List
from db.base import Base

class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(255))
    password: Mapped[str] = mapped_column(String(255))
    user_extension: Mapped["User_Extension"] = relationship(back_populates="user", uselist=False)
    books: Mapped[List["Book"]] = relationship(back_populates="user")


# book.py
from __future__ import annotations
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import Integer, String, Float, ForeignKey
from typing import Optional
from db.base import Base

class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    book_name: Mapped[str] = mapped_column(String(255))
    author: Mapped[str] = mapped_column(String(255))
    price: Mapped[float] = mapped_column(Float)
    publisher: Mapped[str] = mapped_column(String(255))
    
    user_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("user.id"), nullable=True, index=True)
    user: Mapped[Optional["User"]] = relationship(back_populates="books")

关键说明

ForeignKey 的作用

位于“多”方（Book 表），指向“一”方（User.id）。
允许多个 Book 指向同一个 User。
不要加 unique=True，否则会变成一对一。

relationship 配置

端	类型注解	说明
“一”端（User）	`Mapped[List["Book"]]`	返回列表，默认 `uselist=True`
“多”端（Book）	`Mapped["User"]`	返回单个对象

back_populates 实现双向同步，类型注解必须与关系语义一致。

使用示例

1
2
3
4
5
6
7
8


user = User(name="张三", password="xxx")
book1 = Book(book_name="Python入门", author="李四", price=59.9)
book2 = Book(book_name="FastAPI实战", author="王五", price=79.9)

user.books = [book1, book2]  # 自动设置 book1.user_id = user.id

print(book1.user.name)  # "张三"
print(len(user.books))  # 2

三、多对多关系（Many-to-Many）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34


from __future__ import annotations
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import Integer, String, Float, ForeignKey, Table, Column
from typing import List, Optional
from db.base import Base

# 关联表：使用 Table（非 Base 子类）
book_tag = Table(
    "book_tag",
    Base.metadata,
    Column("book_id", Integer, ForeignKey("book.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tag.id"), primary_key=True),
)

class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    book_name: Mapped[str] = mapped_column(String(255))
    author: Mapped[str] = mapped_column(String(255))
    price: Mapped[float] = mapped_column(Float)
    publisher: Mapped[str] = mapped_column(String(255))
    
    user_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("user.id"), nullable=True, index=True)
    user: Mapped[Optional["User"]] = relationship(back_populates="books")
    tags: Mapped[List["Tag"]] = relationship(secondary=book_tag, back_populates="books")


class Tag(Base):
    __tablename__ = "tag"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    tag_name: Mapped[str] = mapped_column(String, unique=True)
    books: Mapped[List["Book"]] = relationship(secondary=book_tag, back_populates="tags")

关键说明

何时使用 Table 或 Base 子类

使用 Table：关联表无额外字段（如 book_tag 只包含 book_id 和 tag_id）。
使用 Base 子类：关联表有额外字段（如 created_at、weight），称为“关联对象模式”。

对于标准多对多，必须使用 Table + Column。原文若使用 mapped_column 是错误的，因为 Table 不接受 MappedColumn。

secondary 参数

指定中间表，可传 Table 对象（推荐）或表名字符串。
SQLAlchemy 自动生成 JOIN 查询。

其他要点

Table 必须使用 Base.metadata，确保 Base.metadata.create_all() 能创建该表。
back_populates 非必需，但建议使用以支持双向导航。

复合主键与多对多关系的逻辑:
在多对多关系中，关联表 book_tag 使用 (book_id, tag_id) 作为复合主键，其作用是确保同一本书和同一个标签的组合只出现一次，而非限制整体为一对一。
每一行代表一个“书-标签”关联。
一本书可关联多个标签：(book1, tag1), (book1, tag2), (book1, tag3) → 合法。
一个标签可被多本书使用：(book1, tag1), (book2, tag1), (book3, tag1) → 合法。
但重复组合如 (book1, tag1) 出现两次 → 违反主键约束不合法因此，复合主键不破坏多对多语义，反而保证了关联的唯一性与一致性，是标准设计
也就是说book_id和tag_id是可重复的，但两者的组合不可重复

使用示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


book = Book(book_name="Python Tricks", author="Dan", price=69.9)
tag1 = Tag(tag_name="Python")
tag2 = Tag(tag_name="Programming")

book.tags = [tag1, tag2]  # 自动插入 book_tag 记录

session.add(book)
await session.commit()

# 查询
book = await session.get(Book, 1)
await session.refresh(book, ["tags"])
print([t.tag_name for t in book.tags])  # ['Python', 'Programming']