跳至內容

零售領域

本導覽說明此示範所使用的 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 是資料表模型,包含 idtimestamp,以及 is_flaggedCustomerSegment 會儲存客群中繼資料。 SegmentPrediction 會傳回 customerIdpredictedSegmentconfidence,以及 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 會使用 customerIdproductCategory,以及 isFlagged

⚠️ SQLModel 會略過 table=True 類別的驗證。因此,設有限制條件的欄位會放在 TransactionBase,並由下列兩者繼承: TransactionTransactionCreate。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()

如果交易已存在,植入方法會立即傳回,因此示範正常重新啟動時具備等冪性。

初始資料

測試資料包含 10 筆交易,涵蓋客戶 C001C005

客戶 初始資料模式
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/segmentsGET /api/segments/{segment_id},以及 GET /api/segments/predict/{customer_id}

聊天端點請參閱 透過 SSE 串流回應,因為這些端點屬於 Copilot 串流路徑,而非零售資料 API。

客群預測邏輯

RetailAnalyticsService.predict_segment 會載入客戶的所有交易,並計算消費總額、平均消費金額與購買頻率,接著依序套用下列規則:

  1. 沒有交易:傳回 New,信賴度為 0.5,並包含 no_history
  2. 消費總額高於 1000:傳回 High Value,信賴度為 0.89
  3. 購買次數達三次以上:傳回 Regular,信賴度為 0.75
  4. 平均消費金額低於 50:傳回 At Risk,信賴度為 0.62
  5. 其他情況:傳回 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. 缺少 null 檢查的位置: 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:

問題原因:臨界值會因市場、季節與零售商而異。檢閱者應將臨界值移至組態或具名原則物件,並在測試中涵蓋邊界行為。