零售领域¶

本演练讲解介绍 Python 零售分析的数据模型、SQLite 配置、种子数据、REST 端点、验证行为，以及演示中有意保留的代码异味。

领域模型¶

API 模型位于 app/models.py。它们使用 SQLModel 实现持久化，并使用 Pydantic 验证请求和响应。

TransactionBase 包含受约束的零售购买字段：

class TransactionBase(SQLModel):
    """Validated fields for a retail purchase transaction."""

    model_config = CAMEL_CONFIG

    customer_id: str = Field(min_length=1, max_length=100)
    amount: float = Field(ge=0.01, le=1_000_000)
    product_category: str = Field(min_length=1, max_length=50)
    store_id: str = Field(min_length=1, max_length=50)

Transaction 是包含 id、timestamp 和 is_flagged 字段的表模型。CustomerSegment 用于存储分段元数据。SegmentPrediction 返回 customerId、predictedSegment、confidence 和 topFeatures。

CamelCase JSON 和验证¶

Python 模型在内部使用 snake_case，但在网络传输时会序列化为 camelCase，以刻意保持 .NET 合约一致性：

#: Serialize as camelCase but still accept snake_case when constructing in Python.
CAMEL_CONFIG = ConfigDict(alias_generator=to_camel, populate_by_name=True)

因此 API JSON 使用 customerId、productCategory 和 isFlagged。

⚠️ SQLModel 会跳过对 table=True 类的验证。受约束字段因此存在于 TransactionBase 中，该基类同时被 Transaction 和 TransactionCreate 继承。FastAPI 在路由调用服务前会验证 TransactionCreate，若请求体无效则返回 HTTP 422 错误：

class TransactionCreate(TransactionBase):
    """Request body for creating a transaction.

    FastAPI validates this automatically and returns HTTP 422 on failure, which
    is what ``ModelState.IsValid`` does in the .NET ``TransactionsController``.
    """

数据库与启动时种子数据¶

app/database.py 创建 SQLite 引擎和每请求会话依赖项：

DATABASE_URL = "sqlite:///retail.db"

engine = create_engine(DATABASE_URL, echo=False)

app/main.py 在 FastAPI 生命周期处理程序中初始化数据：

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Seed database on startup
    create_db_and_tables()
    with Session(engine) as session:
        await RetailAnalyticsService(session).seed_data()

如果交易记录已经存在，种子数据方法会立即返回，因此正常重启演示时该操作具有幂等性。

种子数据¶

该测试数据包含客户 C001 至 C005 的 10 条交易记录：

客户	预置模式
`C001`	杂货和电子产品购买，总额为 335.49
`C002`	杂货和健康产品购买，总额为 47.50
`C003`	电子产品和时尚商品购买，总额为 1,700.00
`C004`	两笔低金额杂货购买，总额为 21.49
`C005`	电子产品和时尚商品购买，总额为 995.00

它还创建了四个客户细分：高价值客户、常规客户、风险客户和新客户。

REST 端点¶

app/routers/transactions.py 提供读取、创建和删除交易的端点：

端点	返回
`GET /api/transactions`	所有 `Transaction` 记录。
`GET /api/transactions/{id}`	一个 `Transaction`，若未找到则返回 `404`。
`POST /api/transactions`	创建交易并返回保存的记录，状态码为 `201 Created`。
`DELETE /api/transactions/{id}`	删除时返回 `204 No Content`，未找到时返回 `404`。

app/routers/segments.py 提供 GET /api/segments、GET /api/segments/{segment_id} 和 GET /api/segments/predict/{customer_id}。

聊天端点在流式传输响应（SSE）中有详细说明，因为它们属于 Copilot 流式传输路径而非零售数据 API。

分段预测逻辑 ¶

RetailAnalyticsService.predict_segment 方法会加载客户的所有交易记录，并计算总消费额、平均消费额及购买频率。随后依次应用以下规则：

无交易记录：返回 New，置信度为 0.5，并包含 no_history。
总消费额超过 1000：返回 High Value，置信度为 0.89。
购买频率达到三次及以上：返回 Regular，置信度为 0.75。
平均消费额低于 50：返回 At Risk，置信度为 0.62。
否则：返回 Regular，置信度为 0.55。

返回的 topFeatures 数组说明了规则输入项，例如总支出、频率或平均支出。

使用 curl 测试¶

真实验证的输出：

$ curl http://localhost:5070/api/segments/predict/C003
{"customerId":"C003","predictedSegment":"High Value","confidence":0.89,"topFeatures":["high_total_spend","multi_category","total_1700"]}

$ curl http://localhost:5070/api/segments/predict/C999
{"customerId":"C999","predictedSegment":"New","confidence":0.5,"topFeatures":["no_history"]}

$ curl http://localhost:5070/api/transactions/1
{"productCategory":"Grocery","customerId":"C001","amount":245.5,"isFlagged":false,"storeId":"S001","id":1,"timestamp":"2026-07-14T03:58:08.543810"}

GET /api/transactions 返回 10 行数据；GET /api/segments 返回 4 行数据；GET /api/transactions/999 返回 HTTP 404。

测试¶

pytest 测试套件与 .NET xUnit 测试保持一致，另包含四个 Python 专用契约测试，用于验证浏览器与 API 之间的请求结构。实际验证结果：

$ uv run pytest
18 passed

四个有意保留的代码异味¶

这些内容是专门为代码评审环节设计的演示材料。请勿将其当作本次演示中需要修复的意外错误；应将其作为评审人员需要发现并解释的示例。它们与 .NET 版本相对应，因此可使用同一份参考答案。

1. N+1 查询问题：`get_transactions_with_segments`¶

问题描述：该方法会加载所有交易记录，然后遍历每个交易记录并调用 predict_segment，后者对每个客户执行一次额外的数据库查询。

for txn in transactions:
    # N+1: querying segments for every single transaction
    segment = await self.predict_segment(txn.customer_id)

问题原因：查询数量随交易记录数量增长。评审人员应建议对客户交易数据进行批处理，或在请求已加载的数据上计算预测结果。

2. `get_transaction` 中缺少空值检查：¶

问题描述：该服务返回了 self._db.get(...) 的结果，即使数据库可能返回无记录。

# Missing null check: will return None if not found
return self._db.get(Transaction, transaction_id)

问题原因：调用方无法从签名中判断 None 是否可能。评审人员应要求提供 Transaction | None 或结果类型，并明确调用方的处理逻辑。

3. 未在 `add_transaction` 中进行输入验证 ¶

问题描述：该服务接收传入的 Transaction，记录时间戳并保存，但未对金额、客户ID、类别或商店值进行校验。

# No validation: negative amounts and empty customer_id are allowed
transaction.timestamp = datetime.now(UTC)
self._db.add(transaction)

问题原因：FastAPI 对 TransactionCreate 进行了验证，但测试、其他服务或未来端点可能直接调用该服务。评审人员应建议将领域不变量集中到服务中，或制定相关政策。

4. `predict_segment` 中硬编码阈值 ¶

问题描述：高价值规则在代码中直接使用字面量阈值 1000。

# BUG: Hardcoded magic number — should be configurable
if total_spend > 1000:

问题所在：阈值会随市场、季节和零售商变化。评审人员应将阈值移至配置或命名策略对象中，并在测试中覆盖边界行为。