db-demo 综合练习

第 16 章 · db-demo 综合练习

本章目标：在 db-demo 完成三项综合交付：纯 SQL 运营报表、SQLAlchemy 批处理脚本、Alembic 一次 schema 迁移；巩固 slug / is_published / price（分） 契约；为 ch17 shop-db 毕业项目热身；对照 ch07～ch11 与 ch13～ch15 优化/cache 文档。

学时建议：5～6 小时（含 3 小时编码）

前置：完成 ch01～ch15；MySQL Docker 与 Python .venv 可用。

16.1 练习背景

db-demo 是 shop-db 的精简沙箱：同一套商品域表，数据量为教学规模（商品 ≥500 行、订单 ≥2000 行 seed）。

db-demo/
├── alembic/
├── db/
│   ├── models.py
│   └── session.py
├── scripts/
│   ├── seed.py
│   ├── report_daily.py      ← 本章 SQLAlchemy
│   └── run_reports.sql      ← 本章纯 SQL
├── sql/
│   └── reports/             ← 报表 SQL 文件
├── docs/
│   ├── PERF.md              ← ch13
│   └── CACHE.md             ← ch15
└── docker-compose.dev.yml

虚构数据：用户 slug 均为 user-demo-*；商品 slug 如 python-handbook；不连接真实生产。

16.2 交付物清单

合计 100 分（本章练习，非毕业答辩）。

16.3 数据模型（复习）

-- 核心表（seed 已存在）
CREATE TABLE products (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    slug VARCHAR(128) NOT NULL UNIQUE,
    name VARCHAR(256) NOT NULL,
    price BIGINT NOT NULL COMMENT '分',
    is_published TINYINT(1) NOT NULL DEFAULT 0,
    category_id BIGINT,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE orders (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(32) NOT NULL UNIQUE,
    user_slug VARCHAR(64) NOT NULL,
    status VARCHAR(16) NOT NULL,
    total_amount BIGINT NOT NULL COMMENT '分',
    created_at DATETIME NOT NULL
);

CREATE TABLE order_items (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_id BIGINT NOT NULL,
    product_id BIGINT NOT NULL,
    product_slug VARCHAR(128) NOT NULL,
    product_name VARCHAR(256) NOT NULL,
    price BIGINT NOT NULL COMMENT '成交单价分',
    quantity INT NOT NULL,
    FOREIGN KEY (order_id) REFERENCES orders(id)
);

冗余字段 product_slug / product_name / price 便于报表 少 JOIN（architecture ch10 快照思想）。

16.4 任务 A：纯 SQL 日报（daily_sales.sql）

需求：统计 昨日 各 订单状态 的订单数、销售额（分）、客单价（分，整数除法）。

sql/reports/daily_sales.sql：

-- 参数：:report_date 例如 '2026-08-18'
SELECT
    o.status,
    COUNT(*) AS order_count,
    SUM(o.total_amount) AS gmv_cents,
    IFNULL(SUM(o.total_amount) / NULLIF(COUNT(*), 0), 0) AS avg_order_cents
FROM orders o
WHERE DATE(o.created_at) = :report_date
GROUP BY o.status
ORDER BY gmv_cents DESC;

16.4.1 运行方式

mysql -h127.0.0.1 -ushop -pshop shop_db \
  -e "SET @report_date='2026-08-18'; SOURCE sql/reports/daily_sales.sql;"

或使用 MySQL 8 用户变量：

SET @report_date = '2026-08-18';
-- 粘贴 SQL，将 :report_date 换为 @report_date
WHERE DATE(o.created_at) = @report_date

16.4.2 验收标准

16.5 任务 B：Top 商品 SQL（top_products.sql）

需求：近 7 日销量 Top 10 商品（按 件数），输出 slug、name、销量、销售额（分）。

SELECT
    oi.product_slug AS slug,
    MAX(oi.product_name) AS name,
    SUM(oi.quantity) AS units_sold,
    SUM(oi.price * oi.quantity) AS revenue_cents
FROM order_items oi
INNER JOIN orders o ON o.id = oi.order_id
WHERE o.created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
  AND o.status IN ('paid', 'shipped', 'completed')
GROUP BY oi.product_slug
ORDER BY units_sold DESC
LIMIT 10;

注意：仅统计已支付相关状态；price * quantity 用 BIGINT，避免 float。

16.6 任务 C：SQLAlchemy 脚本 report_daily.py

需求：Python 脚本串联：

1. 执行 daily_sales + top_products 逻辑（可嵌入 SQL 或 ORM）
2. 输出 output/report-YYYYMMDD.csv
3. 支持 CLI 参数 --date

scripts/report_daily.py：

#!/usr/bin/env python3
"""db-demo 运营日报：SQL + SQLAlchemy 综合"""
from __future__ import annotations

import argparse
import csv
from datetime import date, datetime
from pathlib import Path

from sqlalchemy import create_engine, text

DATABASE_URL = "mysql+pymysql://shop:shop@127.0.0.1:3306/shop_db?charset=utf8mb4"

DAILY_SALES_SQL = text("""
SELECT status, COUNT(*) AS order_count,
       SUM(total_amount) AS gmv_cents
FROM orders
WHERE DATE(created_at) = :report_date
GROUP BY status
""")

TOP_PRODUCTS_SQL = text("""
SELECT oi.product_slug AS slug,
       MAX(oi.product_name) AS name,
       SUM(oi.quantity) AS units_sold,
       SUM(oi.price * oi.quantity) AS revenue_cents
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE DATE(o.created_at) = :report_date
GROUP BY oi.product_slug
ORDER BY units_sold DESC
LIMIT 10
""")


def parse_args():
    p = argparse.ArgumentParser(description="db-demo daily report")
    p.add_argument("--date", default=date.today().isoformat(), help="YYYY-MM-DD")
    p.add_argument("--out-dir", default="output", type=Path)
    return p.parse_args()


def main():
    args = parse_args()
    args.out_dir.mkdir(parents=True, exist_ok=True)
    engine = create_engine(DATABASE_URL, pool_pre_ping=True)

    report_date = args.date
    rows_sales = []
    rows_top = []

    with engine.connect() as conn:
        rows_sales = conn.execute(DAILY_SALES_SQL, {"report_date": report_date}).mappings().all()
        rows_top = conn.execute(TOP_PRODUCTS_SQL, {"report_date": report_date}).mappings().all()

    out_path = args.out_dir / f"report-{report_date.replace('-', '')}.csv"
    with out_path.open("w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["section", "key", "metric", "value"])
        for r in rows_sales:
            w.writerow(["daily_sales", r["status"], "order_count", r["order_count"]])
            w.writerow(["daily_sales", r["status"], "gmv_cents", r["gmv_cents"]])
        for r in rows_top:
            w.writerow(["top_products", r["slug"], "units_sold", r["units_sold"]])
            w.writerow(["top_products", r["slug"], "revenue_cents", r["revenue_cents"]])

    print(f"报告已生成：{out_path}（{datetime.now().isoformat(timespec='seconds')}）")


if __name__ == "__main__":
    main()

16.6.1 ORM 选修：Published 商品计数

from sqlalchemy import select, func
from sqlalchemy.orm import Session
from db.models import Product

def count_published(session: Session) -> int:
    return session.scalar(
        select(func.count()).select_from(Product).where(Product.is_published.is_(True))
    ) or 0

写入 CSV 附加一行 catalog, published_count, ...。

16.7 任务 D：Alembic 一次迁移

需求：为 products 增加 view_count（ BIGINT，默认 0 ），并 回填 seed 随机值（可选）。

16.7.1 修改 models.py

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import BigInteger

class Product(Base):
    __tablename__ = "products"
    # ... slug, price, is_published ...
    view_count: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0, server_default="0")