零售領域¶
本導覽說明此示範所使用的 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 引擎,以及每個要求各自使用的工作階段相依性:
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 筆交易,涵蓋客戶 C001 至 C005:
| 客戶 | 初始資料模式 |
|---|---|
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 的要求結構。實際驗證的測試結果:
四個刻意保留的程式碼異味¶
這些是為程式碼檢閱工作階段刻意準備的示範素材。請勿在示範期間將它們當成意外錯誤並加以修正;應將其作為檢閱者應注意並說明之問題的範例。這些內容與 .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。
問題原因:臨界值會因市場、季節與零售商而異。檢閱者應將臨界值移至組態或具名原則物件,並在測試中涵蓋邊界行為。