一. 前言：資料邊界不是只有「快」或「安全」 #
Python 專案裡，資料通常不是乖乖待在一個地方。 它會從 API 進來，從 queue 進來，從 Redis 快取讀出來，也可能從 JSONL 檔案一行一行掃過去。
這些地方都有同一個問題：你收到的是 bytes 或字串，但程式真正想要的是有結構、有型別、能安心使用的物件。
常見做法通常分成兩派。 一派用 json / orjson 很快地 parse 成 dict，然後自己檢查。 另一派用 Pydantic 建 model，讓 validation 比較清楚。
兩派都很好。 拍拍君不是來拆台的。 但如果你的需求剛好是「decode 很快，而且 decode 的同時做 schema validation」，那就可以看看 msgspec。
msgspec 是一個高效能 serialization / validation 函式庫。 它的核心概念很簡單：
- 用 Python type hints 定義資料 shape。
- 用
msgspec.Struct做輕量資料物件。 - decode JSON 或 MessagePack 時直接轉成 typed object。
- 如果資料格式不對，在邊界就失敗。
如果你看過 Python orjson 實戰，可以把 orjson 想成很快的 JSON 引擎。 如果你看過 Python 資料驗證小幫手：Pydantic，可以把 Pydantic 想成強調 validation 功能與生態整合的 model 工具。
今天這篇不說誰取代誰。 我們只整理一個實務判斷：msgspec 很適合 API client、event payload、cache entry、JSONL ingestion 這種資料邊界。
不要讓髒資料在系統裡旅行太遠。 這就是今天的主題。
二. 安裝：先建立一個乾淨練習專案 #
用 uv 建一個小專案：
mkdir msgspec-demo
cd msgspec-demo
uv init
uv add msgspec
不用 uv 的話：
python -m venv .venv
source .venv/bin/activate
pip install msgspec
確認版本：
import msgspec
print(msgspec.__version__)
這篇會用三個主要模組：
msgspec.Struct：定義資料模型。msgspec.json：JSON encode/decode。msgspec.msgpack：MessagePack encode/decode。
先不急著開 framework。 把資料邊界練清楚，之後放進 FastAPI、CLI、worker、batch job 都比較穩。
三. 第一個 Struct：比 dict 更有邊界 #
先寫一個最小資料模型：
import msgspec
class User(msgspec.Struct):
id: int
name: str
email: str
active: bool = True
建立物件：
user = User(
id=1,
name="拍拍君",
email="ppy@example.com",
)
print(user)
print(user.name)
它看起來有點像 dataclass。 但 msgspec.Struct 是為 serialization 和 validation 設計的。
它有幾個實務上很舒服的特性：
- 欄位固定，不會不小心塞錯 key。
- type hints 直接成為 decode 時的 schema。
- 物件比 dict 更容易補全和重構。
- encode/decode 的效能通常很好。
如果你用 dict，很多錯誤會拖到很後面才爆。 用 Struct 的做法是：資料進來時就檢查。 早點失敗，通常比晚點爆炸便宜很多。
四. JSON decode：從 bytes 直接變成 typed object #
假設 API 回傳一段 JSON bytes：
import msgspec
class User(msgspec.Struct):
id: int
name: str
email: str
active: bool = True
raw = b'''
{
"id": 1,
"name": "拍拍君",
"email": "ppy@example.com"
}
'''
user = msgspec.json.decode(raw, type=User)
print(user)
print(user.active)
重點是這行：
user = msgspec.json.decode(raw, type=User)
它不是先 decode 成 dict，再手動丟進 User(...)。 它是 decode 的時候就知道目標型別。 如果資料不符合 schema，會直接丟錯。
例如 id 變成字串：
bad_raw = b'''
{
"id": "oops",
"name": "拍拍君",
"email": "ppy@example.com"
}
'''
try:
msgspec.json.decode(bad_raw, type=User)
except msgspec.ValidationError as exc:
print(exc)
你會拿到類似這樣的錯誤：
Expected `int`, got `str` - at `$.id`
這種錯誤訊息很適合放在 API client、資料匯入、事件處理器裡。 它直接告訴你哪個欄位錯了、預期型別是什麼、實際拿到什麼。
這比某個很後面的地方發生 KeyError 好處理太多。
五. JSON encode：輸出也可以保持 schema 感 #
decode 是資料進來。 encode 是資料出去。
msgspec 可以把 Struct 直接編成 JSON bytes：
import msgspec
class User(msgspec.Struct):
id: int
name: str
email: str
active: bool = True
user = User(id=1, name="拍拍君", email="ppy@example.com")
raw = msgspec.json.encode(user)
print(type(raw))
print(raw)
輸出是 bytes。 這對很多場景很自然：
- HTTP response body。
- Redis value。
- queue message payload。
- 檔案寫入。
如果你真的需要字串，再 decode：
text = msgspec.json.encode(user).decode("utf-8")
但拍拍君建議先問一句：下一站真的需要 str 嗎？ 很多資料管線其實 bytes 就夠了。
六. MessagePack：同一套型別，換一種 wire format #
JSON 很通用，但不是唯一選擇。 如果你在內部服務、快取、queue、檔案中傳資料，MessagePack 也很常見。
它比 JSON 更緊湊，保留 binary 形態也比較自然。 msgspec 支援同一套 Struct 直接 encode/decode MessagePack。
import msgspec
class Event(msgspec.Struct):
event_id: str
kind: str
user_id: int
payload: dict[str, str]
event = Event(
event_id="evt_001",
kind="user.created",
user_id=1,
payload={"source": "signup"},
)
packed = msgspec.msgpack.encode(event)
decoded = msgspec.msgpack.decode(packed, type=Event)
print(type(packed))
print(decoded.kind)
這對 cache payload 很方便。 你可以把資料用 Struct 表達清楚，再決定 wire format：
- 對外 API：JSON。
- 內部 queue：MessagePack。
- 快取：JSON 或 MessagePack，看除錯和大小需求。
重點是 model 不需要重寫。
七. 實戰：寫一個嚴格 API client #
假設我們要呼叫某個內部 API：
from datetime import datetime
import httpx
import msgspec
class JobStatus(msgspec.Struct, rename="camel"):
job_id: str
state: str
progress: float
updated_at: datetime
client 可以這樣寫：
def fetch_job_status(base_url: str, job_id: str) -> JobStatus:
url = f"{base_url}/jobs/{job_id}"
response = httpx.get(url, timeout=10.0)
response.raise_for_status()
try:
return msgspec.json.decode(response.content, type=JobStatus)
except msgspec.ValidationError as exc:
raise RuntimeError(f"Invalid job status response: {exc}") from exc
這段的邊界很清楚：
- HTTP 錯誤由
raise_for_status()處理。 - schema 錯誤由
msgspec.ValidationError處理。 - function 回傳值保證是
JobStatus。
如果 API response 改版，少了 updatedAt，這段會在 client 邊界立刻爆。 這是好事。 早爆比晚爆好。
八. 測試：schema 錯誤也要測 #
資料邊界的測試不要只測 happy path。 至少補三種：
- 正確 payload 能 decode。
- 少欄位會失敗。
- 型別錯會失敗。
用 pytest 可以這樣寫：
import pytest
import msgspec
class User(msgspec.Struct):
id: int
name: str
email: str
def test_decode_user_ok():
raw = b'{"id": 1, "name": "拍拍君", "email": "ppy@example.com"}'
user = msgspec.json.decode(raw, type=User)
assert user.id == 1
assert user.email == "ppy@example.com"
def test_decode_user_missing_email():
raw = b'{"id": 1, "name": "拍拍君"}'
with pytest.raises(msgspec.ValidationError):
msgspec.json.decode(raw, type=User)
def test_decode_user_wrong_id_type():
raw = b'{"id": "one", "name": "拍拍君", "email": "ppy@example.com"}'
with pytest.raises(msgspec.ValidationError):
msgspec.json.decode(raw, type=User)
如果這段邊界很重要，可以再把 ValidationError 包成你自己的 domain exception。 這樣上層程式不需要知道你底下用的是 msgspec。
九. 遷移策略：不要一口氣全換 #
如果你現在有很多 dict-based code，可以慢慢遷。
第一步，只替最痛的邊界加 schema。 例如 webhook envelope、queue event、cache entry、API response。
第二步，把最常用的 payload 拆成 Struct。 暫時不確定的深層資料，可以先保留成 dict[str, object]，等 shape 穩定後再收斂。
第三步，把 decode 放進單一 function，讓呼叫端只接 typed object。 這樣 msgspec 的使用範圍會集中在邊界，不會散得到處都是。
這樣遷移的好處是風險小、diff 清楚、測試容易補。 不要把整個專案變成大型改造現場。
拍拍君很喜歡這種「先收斂邊界，再收斂內部」的節奏。
十. 什麼時候不要用 msgspec？ #
工具再好，也不是每題都要拿出來。 下面情況可以先不用：
- 資料量很小，標準庫
json已經夠用。 - 團隊已經大量使用 Pydantic，且效能不是瓶頸。
- 需要非常複雜的人類可讀 validation error format。
- model 跟 ORM、表單、OpenAPI schema 生成綁很深。
- 只是一次性腳本，schema 寫完比任務本身還多。
拍拍君的建議是先找一個資料量大、邊界清楚、錯誤代價高的地方試。 例如 API client、event consumer、cache entry。 你會很快感覺到它適不適合你的專案。
不要因為 benchmark 很漂亮就全專案重寫。 效能工具最怕變成架構衝動。
結語 #
msgspec 最迷人的地方，不只是快。 而是它把「快」和「資料契約」放在同一個動作裡。
你可以用 Struct 寫出清楚的 payload shape。 用 msgspec.json.decode(..., type=YourStruct) 在資料進來時驗證。 用 msgspec.json.encode() 或 msgspec.msgpack.encode() 把資料送出去。
今天拍拍君最想你帶走三件事：
msgspec很適合資料邊界，不一定要取代所有 model。- JSON 和 MessagePack 可以共用同一套 typed schema。
- 遷移時先從 API client、event、cache 這些邊界開始。
資料不是只要 parse 成 dict 就結束了。 真正可靠的系統，會在資料剛進門時就問清楚： 你是誰？ 你有沒有缺欄位？ 你是不是我期待的形狀？
問完再放行，後面的程式會輕鬆很多。