毕业项目：svc-demo 微服务 API

第 12 章 · 毕业项目：svc-demo 微服务 API MVP

本章目标：在 ch01～ch11 基础上，独立交付 svc-demo FastAPI 微服务 API 最小可行产品（MVP）；覆盖用户认证、商品 REST API、统一响应与错误码、缓存/限流与运维文档；按 5 天计划实施；通过 100 分验收表自评；完成 curl 冒烟与答辩演示。

学时建议：5 天 × 6～8 小时（合计 30～40 小时）

前置：完成本模块 ch01～ch11；建议复习 python-dev ch06 OOP 与 ch11 httpx/requests；前端联调可参考 frontend-framework API 章节。

12.1 项目背景与边界

svc-demo 是一家虚构的「轻量商品微服务 API 平台」，为 user-demo SPA 与内部运营工具提供 JSON 接口。服务部署在 https://api.example.com（本地等价 http://127.0.0.1:8000）。

┌──────────────────────────────────────────────────────┐
│              svc-demo 服务 MVP（本章）                │
├──────────┬──────────┬──────────┬─────────────────────┤
│ 用户认证  │ 商品 API  │ 横切能力  │  运维与测试          │
│ JWT 登录  │ CRUD+分页 │ 缓存限流  │  pytest / Docker    │
└──────────┴──────────┴──────────┴─────────────────────┘
                        │
                        ▼
              PostgreSQL（教学可用 SQLite）
              Redis（缓存 ch11 + Celery 选修 ch10）

严禁将真实公司内部域名、数据库连接串、密钥写入仓库或答辩材料。统一使用 svc-demo、user-demo、api.example.com。

12.2 技术栈清单

cd ~/python-learn
mkdir svc-demo && cd svc-demo
python -m venv .venv && source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install "fastapi>=0.110" "uvicorn[standard]>=0.27" \
            "sqlalchemy[asyncio]>=2.0" asyncpg alembic \
            "pydantic-settings>=2.0" python-jose[cryptography] passlib[bcrypt] \
            "fastapi-cache2[redis]" slowapi redis httpx pytest pytest-asyncio

12.3 推荐目录结构

svc-demo/
├── .env.example
├── .gitignore
├── docker-compose.dev.yml       # PostgreSQL + Redis
├── alembic.ini
├── pytest.ini
├── requirements.txt
├── docs/
│   ├── API.md                   # 接口文档
│   ├── DEPLOY.md                # 部署说明
│   └── SELF_REVIEW.md           # 100 分自评表
├── scripts/
│   └── smoke.sh                 # curl 冒烟
├── svc_demo/
│   ├── main.py                  # FastAPI app + lifespan
│   ├── core/
│   │   ├── config.py            # BaseSettings
│   │   ├── deps.py              # get_db, get_current_user
│   │   ├── security.py          # JWT
│   │   ├── cache.py
│   │   └── limiter.py
│   ├── models/
│   │   ├── user.py
│   │   └── product.py
│   ├── schemas/
│   │   ├── user.py
│   │   ├── product.py
│   │   └── common.py            # envelope
│   ├── services/
│   │   ├── user.py
│   │   └── product.py
│   ├── api/
│   │   └── v1/
│   │       ├── router.py
│   │       ├── auth.py
│   │       └── products.py
│   └── db/
│       ├── base.py
│       └── session.py
└── tests/
    ├── conftest.py
    ├── test_auth.py
    └── test_products.py

12.4 核心 API 契约

基址：/api/v1

统一响应 envelope

{
  "code": 0,
  "message": "ok",
  "data": { ... }
}

分页列表额外字段：

{
  "pagination": {
    "page": 1,
    "per_page": 10,
    "total": 42,
    "pages": 5
  }
}

错误码约定

12.4.1 模型演进：sku → slug（Alembic 迁移）

ch05～ch06 使用 sku / is_active；毕业项目 ch12 起统一为企业常用字段 slug / is_published。在答辩前执行一次迁移：

cd svc-demo
alembic revision -m "product_sku_to_slug" --autogenerate

迁移脚本要点（手工核对 autogenerate 结果）：

def upgrade():
    op.alter_column("products", "sku", new_column_name="slug")
    op.alter_column("products", "is_active", new_column_name="is_published")

def downgrade():
    op.alter_column("products", "is_published", new_column_name="is_active")
    op.alter_column("products", "slug", new_column_name="sku")

Repository 同步将 get_by_sku 改为 get_by_slug，查询条件改为 is_published.is_(True)。

User 字段与 ch07 对齐

ch07 使用 email + hashed_password（OAuth2 表单 username 传 email）；毕业项目增加 username 列便于 JWT 登录展示：

# 迁移：新增 username，从 email 本地部分回填
op.add_column("users", sa.Column("username", sa.String(64), nullable=True))
# UPDATE users SET username = split_part(email, '@', 1)  # 伪 SQL，按方言调整
op.alter_column("users", "username", nullable=False)

12.5 数据模型要点

User：id, username（唯一）, email, hashed_password, is_active, created_at

Product：id, name, slug（唯一）, price（Numeric）, stock, is_published, description, created_at, updated_at

# svc_demo/models/product.py 片段
from decimal import Decimal
from sqlalchemy import String, Numeric, Boolean, Text
from sqlalchemy.orm import Mapped, mapped_column

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200))
    slug: Mapped[str] = mapped_column(String(200), unique=True, index=True)
    price: Mapped[Decimal] = mapped_column(Numeric(12, 2))
    stock: Mapped[int] = mapped_column(default=0)
    is_published: Mapped[bool] = mapped_column(Boolean, default=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)

答辩时说明：价格用 Decimal/Numeric，避免 float 精度问题。

12.6 种子数据

svc_demo/seed.py（CLI 或 lifespan 首次启动）：

async def seed(session: AsyncSession) -> None:
    if await session.scalar(select(func.count(User.id))):
        return
    user = User(username="admin", email="admin@user-demo.example.com")
    user.hashed_password = hash_password("AdminPass123!")
    session.add(user)
    session.add_all([
        Product(name="FastAPI 实战", slug="fastapi-book", price=Decimal("68.00"),
                stock=50, is_published=True),
        Product(name="异步 Python", slug="async-python", price=Decimal("58.00"),
                stock=30, is_published=True),
    ])
    await session.commit()

# 启动后自动 seed，或
python -m svc_demo.seed