SQLAlchemy ORM 模型与关系 · Python 数据库实战

第 10 章 · SQLAlchemy ORM 模型与关系

本章目标：掌握 SQLAlchemy 2.0 Declarative 风格：Mapped、mapped_column、relationship；实现 一对多（Category→Product、Order→OrderLine）与 多对多（Product↔Tag）；理解 Session 生命周期与 lazy/eager 加载；在 db-demo 建立 ORM 层，字段对齐 slug + is_published + price（分）。

学时建议：5～6 小时（含 2.5 小时模型与关系跟练）

前置：完成 python-database ch09（Core Engine/Connection）；python-dev ch06 面向对象基础。

10.1 场景说明：从 Core Table 到业务对象

ch09 用 Core Table 操作行；业务代码更自然的方式是 Python 类 ↔ 表行 映射：

Category (1) ──► (N) Product (M) ◄──► (N) Tag
     │                    │
     │                    ├── OrderLine ── Order
     │                    └── CartItem ── Cart ── User

关系	ORM 表达	外键
分类 → 商品	`Category.products`	`Product.category_id`
订单 → 明细	`Order.lines`	`OrderLine.order_id`
商品 ↔ 标签	`Product.tags` / `Tag.products`	关联表 `product_tags`
购物车	`Cart.items`	`CartItem.cart_id`

10.2 项目结构与 Base

db-demo/
├── orm/
│   ├── __init__.py
│   ├── base.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── category.py
│   │   ├── product.py
│   │   ├── tag.py
│   │   ├── order.py
│   │   ├── cart.py
│   │   └── user.py
│   └── session.py
└── scripts/demo_orm.py

orm/base.py：

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

10.3 Session 工厂

orm/session.py：

from sqlalchemy.orm import sessionmaker, Session
from core.engine import engine

SessionLocal = sessionmaker(
    bind=engine,
    class_=Session,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
)


def get_session() -> Session:
    return SessionLocal()

参数	说明
`expire_on_commit=False`	commit 后仍可读属性（脚本/序列化友好）
`autoflush=False`	查询前不自动 flush，行为更可预期

典型用法：

from orm.session import get_session

with get_session() as session:
    # SQLAlchemy 2.0 Session 作 context manager：退出时 close
    product = session.get(Product, 1)
    session.commit()

或使用 事务上下文（2.0 推荐）：

with SessionLocal.begin() as session:
    session.add(product)
    # 块结束自动 commit

10.4 Category 与 Product（一对多）

orm/models/category.py：

from __future__ import annotations
from typing import TYPE_CHECKING, List
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, Text
from orm.base import Base

if TYPE_CHECKING:
    from orm.models.product import Product


class Category(Base):
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(64), nullable=False)
    slug: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    sort_order: Mapped[int] = mapped_column(Integer, default=0)

    products: Mapped[List["Product"]] = relationship(
        back_populates="category",
        cascade="save-update, merge",
    )

    def __repr__(self) -> str:
        return f"<Category {self.slug!r}>"

orm/models/product.py：

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

if TYPE_CHECKING:
    from orm.models.category import Category
    from orm.models.tag import Tag

# 多对多关联表
product_tags = Table(
    "product_tags",
    Base.metadata,
    Column("product_id", ForeignKey("products.id", ondelete="CASCADE"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
)


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    category_id: Mapped[int] = mapped_column(ForeignKey("categories.id"), nullable=False)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    slug: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    description: Mapped[Optional[str]] = mapped_column(Text)
    price: Mapped[int] = mapped_column(Integer, nullable=False)           # 分
    stock: Mapped[int] = mapped_column(Integer, default=0)
    is_published: Mapped[bool] = mapped_column(default=False)             # 映射 0/1

    category: Mapped["Category"] = relationship(back_populates="products")
    tags: Mapped[List["Tag"]] = relationship(
        secondary=product_tags,
        back_populates="products",
    )

    @property
    def price_yuan(self) -> str:
        return f"¥{self.price / 100:.2f}"

    def __repr__(self) -> str:
        return f"<Product {self.slug!r} published={self.is_published}>"

要点	说明
`Mapped[int]`	类型注解驱动 2.0 映射
`Mapped[bool]` + SQLite	SQLAlchemy 处理 0/1 与 bool 互转
`price: Mapped[int]`	分，与 ch07 一致
`is_published`	对应 django-web ch07 字段名
`relationship(back_populates=...)`	双向关系

10.5 Tag 与多对多

orm/models/tag.py：

from __future__ import annotations
from typing import TYPE_CHECKING, List
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import String
from orm.base import Base

if TYPE_CHECKING:
    from orm.models.product import Product


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(32), nullable=False)
    slug: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)

    products: Mapped[List["Product"]] = relationship(
        secondary="product_tags",
        back_populates="tags",
    )

操作示例：

with SessionLocal.begin() as session:
    hot = Tag(name="热销", slug="hot")
    mouse = session.scalar(
        select(Product).where(Product.slug == "wireless-mouse")
    )
    mouse.tags.append(hot)
    # commit 时写 product_tags 中间表

M:N 要素	说明
`secondary=`	关联表名或 `Table` 对象
中间表两 FK	复合主键防重复
`append` / `remove`	集合式维护

10.6 Order / OrderLine（一对多级联）

orm/models/order.py：

from __future__ import annotations
from typing import TYPE_CHECKING, List
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import String, Integer, ForeignKey, DateTime, func
from orm.base import Base

if TYPE_CHECKING:
    from orm.models.product import Product