SQLAlchemy 2.0 异步 ORM

第 5 章 · SQLAlchemy 2.0 异步 ORM

本章目标：掌握 SQLAlchemy 2.0 异步栈：create_async_engine、async_sessionmaker、AsyncSession；用 asyncpg 连接 PostgreSQL、用 aiosqlite 做本地练习；定义 Declarative 模型并与 ch03 Pydantic 模型配合（from_attributes）；在 Depends 与 lifespan 中管理会话生命周期；将 svc-demo 商品 API 从内存仓储迁移到真实数据库；对照 Django ORM 与 Flask-SQLAlchemy 说明差异。

学时建议：5～6 小时（含 2 小时跟练）

前置：完成 fastapi-web ch04；了解基本 SQL 与表关系；python-dev 中接触过 SQL 更佳。

5.1 场景说明：svc-demo 需要持久化

ch04 的 ProductMemoryRepo 重启即丢数据。svc-demo 上线到 api.example.com 前必须接入数据库：

5.2 依赖安装

pip install "sqlalchemy[asyncio]>=2.0" aiosqlite asyncpg greenlet

requirements.txt 追加：

sqlalchemy[asyncio]>=2.0,<3
aiosqlite>=0.20          # 本地 SQLite 异步驱动
asyncpg>=0.29             # PostgreSQL 异步驱动（生产）
greenlet>=3.0             # SQLAlchemy 异步必需

5.3 目录结构扩展

svc-demo/
├── app/
│   ├── db/
│   │   ├── __init__.py
│   │   ├── base.py          # DeclarativeBase
│   │   ├── session.py       # engine + sessionmaker
│   │   └── models/
│   │       ├── __init__.py
│   │       ├── category.py
│   │       └── product.py
│   ├── repositories/
│   │   └── product_db.py    # 异步仓储
│   ├── schemas/             # ch03 已有
│   └── main.py              # lifespan 初始化表
├── data/
│   └── svc_demo.db          # SQLite 文件（gitignore）
└── requirements.txt

5.4 引擎与会话 app/db/session.py

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.config import get_settings

settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,       # True 时打印 SQL
    pool_pre_ping=True,        # 连接健康检查
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,    # 提交后仍可访问属性（配合 Pydantic）
    autoflush=False,
    autocommit=False,
)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    """FastAPI Depends 使用的会话依赖"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app/config.py 扩展：

from functools import lru_cache
from pydantic import BaseModel


class Settings(BaseModel):
    app_name: str = "svc-demo"
    api_domain: str = "api.example.com"
    debug: bool = True
    # 练习默认 SQLite；生产改为 postgresql+asyncpg://user:pass@host:5432/svc_demo
    database_url: str = "sqlite+aiosqlite:///./data/svc_demo.db"


@lru_cache
def get_settings() -> Settings:
    return Settings()

5.5 Declarative 模型

5.5.1 基类 app/db/base.py

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

5.5.2 分类模型 app/db/models/category.py

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

from app.db.base import Base


class Category(Base):
    __tablename__ = "categories"

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

    products: Mapped[list["Product"]] = relationship(back_populates="category")

5.5.3 商品模型 app/db/models/product.py

from datetime import datetime
from decimal import Decimal

from sqlalchemy import Boolean, DateTime, ForeignKey, Numeric, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db.base import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    sku: Mapped[str] = mapped_column(String(32), unique=True, nullable=False, index=True)
    name: Mapped[str] = mapped_column(String(128), nullable=False)
    price: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    stock: Mapped[int] = mapped_column(default=0, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    category_id: Mapped[int] = mapped_column(ForeignKey("categories.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )

    category: Mapped["Category"] = relationship(back_populates="products")