一. 前言：銀牌竟然排在金牌後面？ #
嗨，我是拍拍君。今天替客服工單整理會員等級。
報表想依「銅牌、銀牌、金牌」排序，程式卻排成 bronze、gold、silver。
隔天另一批又冒出 Gold、空字串和 platinum，大家都說自己沒改 schema。
欄位仍叫 tier，型別仍是字串，資料的意思卻已經變了。
這次不是換一個比較省 RAM 的 dtype 就收工，
而是約定：哪些詞可以進來、未知值去哪裡、排序按什麼規則、下一批如何相容。
需要 expression 基礎可先看 Polars 入門。
這裡的 pl.Enum 是整欄資料的 dtype，
不是 Python enum 類別，也不負責驗證狀態轉換。
二. 安裝與檔案安排 #
本文在 Python 3.12、Polars 2.1.0 實測，先固定版本：
uv init pypy-category-contracts
cd pypy-category-contracts
uv add 'polars==2.1.0'
uv add --dev 'pytest==9.2.0'
第四節存成 category_contract.py；其餘示範 Python 區塊依序接進 demo.py。
第十節另外存成 test_category_contract.py。
範例是能放進記憶體的小表，沒有聲稱適用所有資料規模。
先把三種需求分開：
| 欄位需求 | 選擇 | 不會自動做到的事 |
|---|---|---|
| 已核准且有順序的有限詞彙 | pl.Enum |
不代表欄位不可空 |
| 新分類本來就能隨來源增加 | pl.Categorical |
不會替你拒絕拼字錯誤 |
| 開放文字或幾乎每列都不同 | pl.String |
沒有分類詞彙限制 |
Enum 的宣告順序是契約的一部分，不是純粹顯示用的清單。
Categorical 則是編碼選擇，不等於已核准的分類目錄。
API 語意可對照 官方分類資料指南。
三. 最小對照：固定詞彙不等於禁止 Null #
先在 demo.py 試兩件事：未知字串會失敗，原本的 Null 可以保留。
這裡把預期錯誤接住，不讓整份示範中途停止。
import polars as pl
TIER = pl.Enum(["bronze", "silver", "gold"])
s = pl.Series("tier", ["gold", "bronze", None], dtype=TIER)
assert s.null_count() == 1
try:
pl.Series("tier", ["platinum"], dtype=TIER)
except pl.exceptions.InvalidOperationError:
pass
else:
raise AssertionError("未知等級竟然通過")
raw = pl.Series("tier", ["gold", "platinum", None])
lossy = raw.cast(TIER, strict=False)
assert lossy.to_list() == ["gold", None, None]
最後一行看起來很方便，卻把「未知 platinum」和「來源本來缺值」混在一起。
strict=False 不是自動修好資料，而是容許轉型失敗變 Null。
若下游再 drop_nulls()，一筆原本可追查的異常就悄悄消失了。
本篇政策是：tier 不可空、大小寫精確、空白也不能自動修剪。
如果你的來源允許 GOLD，應另外定義正規化規則並保留原始值；
不要為了讓 fixture 通過，臨時加上 strip().lower()。
轉型細節見 Expr.cast API。
四. 入口契約：先分類原因，再轉成 Enum #
以下整段存成 category_contract.py。
它只接受工單 ID 與原始 tier 兩欄，ID 必須唯一且不可空。
分區函式可以產生診斷結果；正式入口預設遇到拒收資料就失敗。
import polars as pl
VOCAB_VERSION = "tier-v1"
LABELS = ["bronze", "silver", "gold"]
TIER = pl.Enum(LABELS)
INPUT_SCHEMA = {"ticket_id": pl.Int64, "tier": pl.String}
class ContractError(ValueError):
pass
def partition_tiers(frame):
if dict(frame.schema) != INPUT_SCHEMA:
raise ContractError("input_schema")
ids = frame["ticket_id"]
if ids.null_count() or ids.n_unique() != frame.height:
raise ContractError("ticket_identity")
marked = frame.rename({"tier": "tier_raw"}).with_columns(
pl.when(pl.col("tier_raw").is_null()).then(pl.lit("missing"))
.when(pl.col("tier_raw").is_in(LABELS)).then(pl.lit("accepted"))
.otherwise(pl.lit("unknown")).alias("admission")
)
accepted = marked.filter(pl.col("admission") == "accepted").with_columns(
pl.col("tier_raw").cast(TIER, strict=True).alias("tier")
)
rejected = marked.filter(pl.col("admission") != "accepted")
if accepted.height + rejected.height != frame.height:
raise ContractError("reconciliation")
return accepted, rejected
def admit_tiers(frame):
accepted, rejected = partition_tiers(frame)
if not rejected.is_empty():
raise ContractError("unapproved_tier")
return accepted
判斷 Null 放在最前面，才能穩定產生 missing 原因。
未知值保留 tier_raw，不把它補成金牌，也不把它改成合法等級 unknown。
後者若是產品真實需求，必須先把 unknown 納入詞彙與語意，而非偷偷放行。
這裡的 ID 唯一只涵蓋單次輸入，不保證兩個批次沒有重複工單。 正式跨批次處理仍要在合併後檢查 ID，或由儲存層負責唯一約束。 同樣，這個契約不驗證會員資格、工單權限或事件發生時間。
五. 隔離資料：報表不能只剩下好看的列 #
回到 demo.py，準備一批混合資料：
from category_contract import (
ContractError, INPUT_SCHEMA, LABELS, TIER, VOCAB_VERSION,
admit_tiers, partition_tiers,
)
raw = pl.DataFrame({
"ticket_id": [1, 2, 3, 4, 5, 6],
"tier": ["gold", "bronze", "silver", "platinum", None, "Gold"],
}, schema=INPUT_SCHEMA)
accepted, rejected = partition_tiers(raw)
assert accepted["ticket_id"].to_list() == [1, 2, 3]
assert rejected["tier_raw"].to_list() == ["platinum", None, "Gold"]
assert rejected["admission"].to_list() == ["unknown", "missing", "unknown"]
assert accepted.height + rejected.height == raw.height == 6
try:
admit_tiers(raw)
except ContractError as exc:
assert str(exc) == "unapproved_tier"
else:
raise AssertionError("正式入口不應忽略拒收列")
本批接收 3/6，未知 2/6，缺值 1/6；分母都是同一批六筆。 診斷輸出可以給維護者修來源，但不能自動當成已完整發布的報表。 正式修復應留下 batch ID、詞彙版本與拒收數量，重新處理同一份來源快照。 本篇範例只含虛構等級；真實隔離區也需要適當的存取與保留政策。
六. 排序：業務等級不是字典序 #
同一組標籤，轉成兩種 dtype，排序就有不同意思：
enum_sorted = accepted.sort(["tier", "ticket_id"])
assert enum_sorted["tier"].to_list() == ["bronze", "silver", "gold"]
cat = accepted.with_columns(pl.col("tier_raw").cast(pl.Categorical).alias("tier"))
assert cat.sort("tier")["tier"].to_list() == ["bronze", "gold", "silver"]
assert accepted.filter(pl.col("tier") >= "silver")["ticket_id"].to_list() == [1, 3]
在本次測試版本中，Enum 依宣告順序；Categorical 排序用字典序。 所以「至少銀牌」必須在有業務順序的欄位上運算，不能拿文字排序代替。 同等級再按 ticket_id 排，則是為了明確決定平手順序。 別把某次 group_by 恰巧輸出的順序當成報表保證。
七. 跨批次：共用 dtype，不共用猜測 #
已知等級不必每批看見全部三種，但應使用同一個完整的 TIER：
batch_a = admit_tiers(pl.DataFrame({
"ticket_id": [10], "tier": ["bronze"],
}, schema=INPUT_SCHEMA))
batch_b = admit_tiers(pl.DataFrame({
"ticket_id": [11], "tier": ["gold"],
}, schema=INPUT_SCHEMA))
combined = pl.concat([batch_a, batch_b], how="vertical")
assert combined["tier"].dtype == TIER
assert combined["tier"].to_list() == ["bronze", "gold"]
assert combined["ticket_id"].n_unique() == combined.height
reversed_tier = pl.Enum(list(reversed(LABELS)))
assert reversed_tier != TIER
try:
pl.concat([batch_a["tier"], batch_b["tier"].cast(reversed_tier)])
except pl.exceptions.SchemaError:
pass
else:
raise AssertionError("不同排序契約不應直接混用")
每批 unique() 後自己建 Enum，很容易做出不同詞彙或不同順序。
兩批都顯示 enum，不代表完整 dtype 相同。
也不要用 relaxed concat 把差異藏起來，先決定誰才是正式詞彙。
ID 衝突與分類相容是兩個獨立檢查；Join 關係則請看 基數契約。
八. 開放分類：Categorical 與 Categories 的邊界 #
來源通路若允許增加新名稱，就不必硬塞進固定 Enum。 下面用具名 mapping 展示兩個獨立建立的批次；不是建立 allowlist。
channel_type = pl.Categorical(pl.Categories("pypy_channels", namespace="demo"))
channels_a = pl.Series("channel", ["web", "chat"], dtype=channel_type)
channels_b = pl.Series("channel", ["chat", "kiosk"], dtype=pl.Categorical(
pl.Categories("pypy_channels", namespace="demo")
))
channels = pl.concat([channels_a, channels_b])
assert channels.to_list() == ["web", "chat", "chat", "kiosk"]
assert channels.n_unique() == 3
kiosk 被接收正是本例想要的行為；如果是拼錯的 kioskk，一樣會被編碼。
在 Polars 2.1.0，預設 Categorical 使用全域 mapping；
具名 Categories 的相容性取決於 name、namespace 與 physical type。
它是同一行程內的編碼協調方式，不是跨服務永久 ID 登記系統。
安裝版文件把 Categories 標成 unstable，升版前應重跑相容性測試。
不要把 to_physical() 的數字輸出當成客戶等級或通路主鍵。
交換資料應傳標籤與詞彙版本，必要時在接收端重新轉型。
舊教學的 with pl.StringCache(): 也不該照搬：
目前 StringCache 文件
已註明它自 1.41.0 起 deprecated，且現在不做任何事。
九. 詞彙升版與檔案往返 #
假設產品核准新增白金牌，不要在 v1 背後悄悄改清單。 建立 v2、重新驗證舊資料，並讓接收端明確知道新的排序規則：
TIER_V2 = pl.Enum([*LABELS, "platinum"])
migrated = combined.with_columns(
pl.col("tier").cast(pl.String).cast(TIER_V2, strict=True)
)
assert migrated["tier"].dtype == TIER_V2
assert migrated["tier"].cast(pl.String).to_list() == ["bronze", "gold"]
刪除或重新命名分類更需要業務轉換表，不能期待 cast 猜到新意思。 即使只重排清單，比較與排序也會改變，應視為語意變更。 以下使用暫存目錄，驗證同一版本 Polars 的 Parquet 與 CSV 往返：
import json
from pathlib import Path
from tempfile import TemporaryDirectory
from polars.testing import assert_frame_equal
with TemporaryDirectory() as folder:
root = Path(folder)
combined.write_parquet(root / "tickets.parquet")
restored = pl.read_parquet(root / "tickets.parquet")
assert_frame_equal(restored, combined)
assert restored["tier"].dtype == TIER
manifest = {"vocabulary_version": VOCAB_VERSION, "labels": LABELS}
(root / "vocabulary.json").write_text(json.dumps(manifest), encoding="utf-8")
combined.select("ticket_id", "tier").write_csv(root / "tickets.csv")
loaded = json.loads((root / "vocabulary.json").read_text(encoding="utf-8"))
if loaded != {"vocabulary_version": VOCAB_VERSION, "labels": LABELS}:
raise ContractError("vocabulary_manifest")
csv_input = pl.read_csv(root / "tickets.csv", schema_overrides=INPUT_SCHEMA)
rebuilt = admit_tiers(csv_input)
assert_frame_equal(rebuilt, combined)
這證明的是本版本原生 Polars writer/reader 的行為，不是所有工具都保留 Enum。 CSV 不攜帶這份 dtype，必須重新套用契約；JSON 要像上面一樣明確載入與比對。 真實交換時需驗證 manifest 版本與有序標籤，不能只把檔案放旁邊就算完成。 外部工具改寫 Parquet 後，應再檢查 schema、值與排序語意。 這與 PyArrow 資料交換 的一般檔案讀寫不同，驗收對象是分類契約。
十. 回歸測試：空表、拼字錯誤和 Null 都要測 #
以下存成 test_category_contract.py，不只測 happy path：
import polars as pl
import pytest
from category_contract import (
ContractError, INPUT_SCHEMA, TIER, admit_tiers, partition_tiers,
)
@pytest.mark.parametrize("value,reason", [
("platinum", "unknown"), ("Gold", "unknown"),
("", "unknown"), (" gold ", "unknown"), (None, "missing"),
])
def test_rejection_retains_raw(value, reason):
frame = pl.DataFrame({"ticket_id": [1], "tier": [value]}, schema=INPUT_SCHEMA)
good, bad = partition_tiers(frame)
assert good.is_empty() and good["tier"].dtype == TIER
assert bad["tier_raw"].item() == value
assert bad["admission"].item() == reason
with pytest.raises(ContractError, match="^unapproved_tier$"):
admit_tiers(frame)
def test_empty_input_keeps_enum():
result = admit_tiers(pl.DataFrame(schema=INPUT_SCHEMA))
assert result.is_empty() and result["tier"].dtype == TIER
@pytest.mark.parametrize("ids", [[1, 1], [1, None]])
def test_bad_identity_fails(ids):
frame = pl.DataFrame({"ticket_id": ids, "tier": ["gold", "silver"]},
schema=INPUT_SCHEMA)
with pytest.raises(ContractError, match="^ticket_identity$"):
partition_tiers(frame)
uv run python demo.py
uv run pytest -q
示範的 assert 用來核對預期；正式入口則明確 raise，不受 python -O 移除影響。
上線前再加來源 schema 改變、跨批次 ID 重複、詞彙刪除與外部工具改寫案例。
若只量效能，至少也要確認轉型前後的值、Null、拒收數量一致。
不要因為「分類欄通常比較省」就承諾每份資料都會加速。
結語：先管詞彙，再談編碼 #
拍拍君的選擇很簡單：核准過且有順序的詞彙用 Enum， 允許自然增加的分類才用 Categorical，開放文字則保留 String。 未知值與 Null 分開追蹤，同一批資料完整對帳，再決定能不能發布。 跨批次共用明確 dtype，跨檔案驗證標籤與詞彙版本，才不會「看起來都是 enum」卻各說各話。