Alembic 迁移与 CRUD 仓库模式

第 6 章 · Alembic 迁移与 CRUD 仓库模式

本章目标：在 svc-demo 中配置 Alembic 异步迁移；用 Repository 模式 封装 Product 数据访问；实现完整的 Product CRUD API（创建、查询、更新、软删除）；理解迁移版本链与回滚；为 ch07 JWT 保护写接口打基础。

学时建议：5～6 小时（含 2 小时迁移与 CRUD 跟练）

前置：完成 fastapi-web ch01～ch05（FastAPI 项目骨架、AsyncSession、ch05 的 Product/Category ORM 模型与 get_db 依赖）。

字段说明：ch05～ch06 商品模型使用 sku + is_active；ch12 毕业项目演进为 slug + is_published（见 §12.4.1 Alembic 迁移）。ch06 本章 Repository/API 与 ch05 字段保持一致。

6.1 场景说明：从「能连库」到「能运营」

ch05 已用 SQLAlchemy 2.0 异步 ORM 定义模型并手工建表。真实项目需要：

痛点	方案	本章技术
表结构随版本演进	可追踪的 DDL 脚本	Alembic `revision` / `upgrade`
路由里堆满 SQL	分层解耦	Repository 模式
商品增删改查重复代码	统一接口	`ProductRepository` + Pydantic Schema
多人协作冲突	版本号链	`alembic_version` 表

教学项目 svc-demo 本地运行于 http://127.0.0.1:8000；对外文档域名使用虚构 https://api.example.com，不涉及任何企业内部仓库或配置。

6.2 安装 Alembic 与初始化

在 svc-demo 虚拟环境中：

cd ~/python-learn/svc-demo
pip install "alembic>=1.13,<2"
pip freeze | grep -i alembic >> requirements.txt
alembic init alembic

生成目录结构：

svc-demo/
├── alembic/
│   ├── versions/          # 迁移脚本
│   ├── env.py             # 迁移运行时入口
│   └── script.py.mako
├── alembic.ini
└── app/
    ├── core/
    │   └── database.py    # ch05 已有
    └── models/

6.3 配置异步 env.py

编辑 alembic/env.py，接入 ch05 的 Base 与异步引擎。核心要点：

from app.core.config import settings
from app.core.database import Base
from app.models import product, category  # noqa: F401

config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)
target_metadata = Base.metadata

async def run_async_migrations() -> None:
    connectable = async_engine_from_config(..., poolclass=pool.NullPool)
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)  # 在 async 连接上跑同步迁移
    await connectable.dispose()

def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())

完整模板见 Alembic 官方文档；offline 模式保留默认 run_migrations_offline() 即可。

alembic.ini 中 sqlalchemy.url 可留空，由 env.py 从 settings 注入。

配置点	说明
`target_metadata`	绑定 ORM `Base.metadata`，自动生成 DDL
`run_sync`	在异步连接上执行同步迁移逻辑
模型 import	必须 import 所有模型，否则 autogenerate 漏表

6.4 首次迁移与常用命令

# 自动生成迁移（对比模型与数据库）
alembic revision --autogenerate -m "init product category"

# 升级到最新
alembic upgrade head

# 查看当前版本
alembic current

# 回滚一步
alembic downgrade -1

# 查看历史
alembic history --verbose

迁移脚本审查要点（alembic/versions/xxxx_init.py）：

upgrade() 与 downgrade() 必须成对
重命名列用 op.alter_column，不要 drop + add（丢数据）
生产环境先 upgrade 到 staging，再上线

autogenerate 产出后重点检查：create_table 字段类型、ForeignKey、server_default 是否与模型一致；downgrade() 必须可逆。

6.5 Repository 模式：为什么需要

直接在路由里写 session.execute(select(Product)...) 会导致：

路由臃肿，难以单测
查询逻辑散落，改一处漏多处
换存储实现（如加缓存层）成本高

分层示意：

HTTP 请求
  → API Router（参数校验、状态码）
  → Service（可选，复杂业务编排）
  → Repository（纯数据访问）
  → AsyncSession → PostgreSQL

本章在 Repository 层完成 CRUD；ch07 起在 Router 层加认证依赖。

6.6 ProductRepository 实现

app/repositories/product.py：

from decimal import Decimal
from typing import Sequence

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.product import Product


class ProductRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get_by_id(self, product_id: int) -> Product | None:
        stmt = (
            select(Product)
            .options(selectinload(Product.category))
            .where(Product.id == product_id, Product.is_active.is_(True))
        )
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def get_by_sku(self, sku: str) -> Product | None:
        stmt = select(Product).where(Product.sku == sku, Product.is_active.is_(True))
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def list(
        self,
        *,
        skip: int = 0,
        limit: int = 20,
        category_id: int | None = None,
        keyword: str | None = None,
    ) -> tuple[Sequence[Product], int]:
        base = select(Product).where(Product.is_active.is_(True))
        count_stmt = select(func.count()).select_from(Product).where(Product.is_active.is_(True))

        if category_id is not None:
            base = base.where(Product.category_id == category_id)
            count_stmt = count_stmt.where(Product.category_id == category_id)
        if keyword:
            like = f"%{keyword}%"
            base = base.where(Product.name.ilike(like))
            count_stmt = count_stmt.where(Product.name.ilike(like))

        total = (await self.session.execute(count_stmt)).scalar_one()
        stmt = (
            base.options(selectinload(Product.category))
            .order_by(Product.id.desc())
            .offset(skip)
            .limit(limit)
        )
        rows = (await self.session.execute(stmt)).scalars().all()
        return rows, total

    async def create(self, **fields) -> Product:
        product = Product(**fields)