Python JSON 實戰：解析、序列化、自訂型別與大型資料

Python 學習 - 本文屬於一個選集。

§ 119: 本文

一. 前言：JSON 不是「把 dict 存起來」而已
#

JSON 幾乎無所不在。 API 回應、設定檔、事件紀錄、CLI 輸出，甚至模型產生的結構化結果，都可能是一段 JSON。第一版通常只是 json.loads(raw_text)，可是資料一離開自己的電腦，麻煩就來了：

中文被轉成一排 \uXXXX
金額經過 float 後出現精度誤差
NaN 被寫進檔案，別的系統卻拒絕
重複 key 悄悄覆蓋前面的值
自訂型別無法序列化
巨大陣列一次載入，記憶體直接暴衝
多份 JSON 直接連寫，下一次再也讀不回來 Python 的 json 是標準庫，不用安裝，也不代表只能做玩具範例。它提供 encoder、decoder hooks、精度控制、CLI 驗證與串流寫出等實用能力。這篇專注在「格式邊界怎麼設計」。如果你需要高吞吐量與 bytes 輸出，請看 orjson 實戰；如果想在 decode 時直接套用型別 schema，可以接著看 msgspec 實戰。

二. 準備：標準庫內建，先分清四個函式
#

不需要 pip install。建立一個練習目錄即可：

mkdir pypy-json-lab
cd pypy-json-lab
uv init

json 最常用的 API 可以分成兩組：

API	輸入／輸出	適合情境
`json.dumps(obj)`	Python 物件 → `str`	API body、log、測試
`json.loads(text)`	`str`/`bytes` → Python 物件	解析收到的內容
`json.dump(obj, fp)`	Python 物件 → 文字檔	直接寫檔
`json.load(fp)`	文字檔 → Python 物件	直接讀檔
多一個 `s`，可以記成 string。
先看最小 round trip：

import json
order = {
    "id": 1001,
    "customer": "拍拍君",
    "paid": True,
    "items": ["keyboard", "coffee"],
    "coupon": None,
}
text = json.dumps(order, ensure_ascii=False)
restored = json.loads(text)
assert restored == order
print(text)

JSON object、array、string 會分別成為 Python 的 dict、list、str；數字、布林與 null 則對應 int/float、bool 與 None。注意：JSON object 的 key 必須是字串。 Python dict 若用整數 key，序列化後會變成字串，因此 loads(dumps(data)) 不一定等於原物件。

三. 輸出策略：人類可讀、機器緊湊、結果穩定
#

寫設定檔時，排版比少幾個 byte 重要：

pretty = json.dumps(
    order,
    ensure_ascii=False,
    indent=2,
    sort_keys=True,
)
print(pretty)

ensure_ascii=False 會保留中文，而不是全部跳脫成 ASCII。寫入檔案時要明確指定 UTF-8：

from pathlib import Path
import json
path = Path("order.json")
path.write_text(
    json.dumps(order, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)

若是網路傳輸或 cache，通常使用緊湊格式：

compact = json.dumps(
    order,
    ensure_ascii=False,
    separators=(",", ":"),
)

sort_keys=True 可以讓測試 fixture、diff 與快取 key 更穩定。但它不是完整的 JSON canonicalization 規格；涉及簽章或跨語言雜湊時，請採用雙方明確同意的 canonical format，別把「看起來固定」當成密碼學保證。

四. `dump()` 與檔案：一個檔案只放一份 JSON document
#

json.dump() 直接把內容寫到支援 .write() 的文字 file object：

from pathlib import Path
import json
path = Path("order.json")
with path.open("w", encoding="utf-8") as file:
    json.dump(order, file, ensure_ascii=False, indent=2)
    file.write("\n")

讀回來則使用 load()：

with path.open("r", encoding="utf-8") as file:
    loaded = json.load(file)
assert loaded["id"] == 1001

不要對同一個檔案連續呼叫兩次 dump()：

# 錯誤示範：結果是兩份黏在一起的 JSON，並非合法單一 document
with open("broken.json", "w", encoding="utf-8") as file:
    json.dump({"id": 1}, file)
    json.dump({"id": 2}, file)

JSON 本身不是 framed protocol，不知道第一份何時結束、第二份何時開始。要儲存多筆紀錄，可以用一個 JSON array；資料會持續增加時，則改用後面會介紹的 JSON Lines。

五. 自訂序列化：用 `default` 建立明確協定
#

datetime、Decimal、Path 等 Python 物件不是原生 JSON 型別。直接序列化會得到 TypeError：

from datetime import datetime, timezone
import json
event = {
    "name": "deploy",
    "created_at": datetime.now(timezone.utc),
}
# json.dumps(event)  # TypeError

可以把轉換規則集中在 default：

from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any
import json
def encode_extra(value: Any) -> Any:
    if isinstance(value, datetime):
        return {"__type__": "datetime", "value": value.isoformat()}
    if isinstance(value, date):
        return {"__type__": "date", "value": value.isoformat()}
    if isinstance(value, Decimal):
        return {"__type__": "decimal", "value": str(value)}
    if isinstance(value, Path):
        return {"__type__": "path", "value": str(value)}
    raise TypeError(f"unsupported type: {type(value).__name__}")
text = json.dumps(event, default=encode_extra, ensure_ascii=False)

最後一定要 raise TypeError。若偷偷 return str(value)，未處理的型別也會被吞掉，錯誤只會換一種更難追的形式出現。簡單專案用 default 函式就夠了；需要重用複雜設定或包成 framework adapter 時，也可繼承 JSONEncoder 並覆寫同名方法。

六. 自訂解析：`object_hook` 只做可辨識的轉換
#

剛才輸出的 tagged object 可以用 object_hook 還原：

from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any
import json
def decode_extra(obj: dict[str, Any]) -> Any:
    kind = obj.get("__type__")
    value = obj.get("value")
    if kind == "datetime" and isinstance(value, str):
        return datetime.fromisoformat(value)
    if kind == "date" and isinstance(value, str):
        return date.fromisoformat(value)
    if kind == "decimal" and isinstance(value, str):
        return Decimal(value)
    if kind == "path" and isinstance(value, str):
        return Path(value)
    return obj
restored = json.loads(text, object_hook=decode_extra)

hook 會對每個 JSON object 呼叫，包含巢狀 object。因此 tag 必須夠明確，且未命中時原樣回傳 dict。也別把 object_hook 當成完整 schema validator。欄位是否缺少、字串長度是否合理、值是否符合商業規則，應該在解析之後明確驗證；需要模型化驗證時，可參考 Pydantic 教學。

七. 金額與精度：解析成 `Decimal`
#

JSON 只有一種 number 語法，Python 預設把帶小數點的數字解析成 float：

import json
data = json.loads('{"price": 0.1, "tax": 0.2}')
print(data["price"] + data["tax"])
# 0.30000000000000004

金額或需要十進位精度的資料，可以指定 parse_float：

from decimal import Decimal
import json
data = json.loads(
    '{"price": 0.1, "tax": 0.2}',
    parse_float=Decimal,
)
assert data["price"] + data["tax"] == Decimal("0.3")

parse_int 也能替換整數解析器，但不要隨便繞過 Python 對超長整數字串的安全限制。還有一個跨語言陷阱：Python int 可以非常大，許多 JSON 消費端卻用 IEEE 754 double。例如資料庫 ID 超過 JavaScript 的安全整數範圍時，最好在 API 契約裡傳成字串，而不是期待每個消費端都保留精度。

八. 嚴格模式：拒絕 NaN、Infinity 與重複 key
#

RFC 8259 不允許 NaN 與 Infinity。 Python 為了相容性，預設卻可以輸出它們：

import json
print(json.dumps({"score": float("nan")}))
# {"score": NaN}

對外交換資料時，建議主動拒絕：

json.dumps(
    {"score": float("nan")},
    allow_nan=False,
)
# ValueError

解析端則可用 parse_constant：

def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite number is forbidden: {value}")
json.loads('{"score": NaN}', parse_constant=reject_constant)

重複 key 也要小心：

raw = '{"role": "reader", "role": "admin"}'
print(json.loads(raw))
# {'role': 'admin'}

預設只留下最後一個值。需要拒絕重複名稱時，使用 object_pairs_hook：

from typing import Any
import json
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate key: {key}")
        result[key] = value
    return result
json.loads(raw, object_pairs_hook=unique_object)

object_pairs_hook 比 object_hook 優先，因為只有前者看得到 key 被合併前的順序與重複項目。

九. 錯誤處理：把位置資訊留給人看
#

格式錯誤時會拋出 JSONDecodeError：

import json
raw = '{"name": "拍拍君", "active": tru}'
try:
    json.loads(raw)
except json.JSONDecodeError as error:
    print(error.msg)
    print(f"line={error.lineno}, column={error.colno}, pos={error.pos}")

在 CLI 或匯入工具裡，請回報行、列與來源檔名。不要把完整 payload 直接寫進 log，因為內容可能包含 token、email 或其他敏感資料。命令列也能快速檢查與 pretty-print：

python -m json order.json

這很適合 pre-commit、CI 或臨時檢查，不必另外安裝工具。

十. JSON Lines：讓大量紀錄可以一行一筆
#

一個巨大 JSON array 必須先解析整份文件，才能取得裡面的 list。若資料天然是一筆筆獨立紀錄，可改用 JSON Lines，也常寫成 .jsonl 或 NDJSON。每一行都是完整 JSON value：

from collections.abc import Iterable
from pathlib import Path
from typing import Any
import json
def write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None:
    with path.open("w", encoding="utf-8") as file:
        for row in rows:
            text = json.dumps(
                row,
                ensure_ascii=False,
                separators=(",", ":"),
                allow_nan=False,
            )
            file.write(text + "\n")

讀取時逐行解析：

from collections.abc import Iterator
def read_jsonl(path: Path) -> Iterator[dict[str, Any]]:
    with path.open("r", encoding="utf-8") as file:
        for line_number, line in enumerate(file, start=1):
            if not line.strip():
                continue
            try:
                value = json.loads(
                    line,
                    parse_float=Decimal,
                    object_pairs_hook=unique_object,
                    parse_constant=reject_constant,
                )
            except (json.JSONDecodeError, ValueError) as error:
                raise ValueError(f"line {line_number}: {error}") from error
            if not isinstance(value, dict):
                raise TypeError(f"line {line_number}: object required")
            yield value

這種流程的記憶體大致跟「單行最大紀錄」有關，而不是跟整個檔案大小一起成長。它也方便 append、斷點續跑與逐筆錯誤回報。但一筆紀錄仍然不能任意巨大；如果單行本身有數百 MB，照樣會吃掉大量記憶體。

十一. 不可信輸入：先限制大小，再呼叫 parser
#

官方文件特別提醒：惡意 JSON 可能消耗大量 CPU 與記憶體。 json.loads() 沒有 max_bytes= 參數，因此限制要放在資料進入 parser 之前。讀 HTTP response 時先檢查 header 還不夠，因為 Content-Length 可能缺少或不可信；實務上要讓 HTTP client 以串流方式讀取並累計 byte 數。本機檔案可以先檢查大小：

from pathlib import Path
from typing import Any
import json
def load_small_json(path: Path, max_bytes: int = 2_000_000) -> Any:
    size = path.stat().st_size
    if size > max_bytes:
        raise ValueError(f"JSON file too large: {size} bytes")
    with path.open("r", encoding="utf-8") as file:
        return json.load(
            file,
            parse_constant=reject_constant,
            object_pairs_hook=unique_object,
        )

還要依系統風險限制：

HTTP request body 大小
壓縮後與解壓後大小
巢狀深度
單一字串與單筆紀錄長度
處理時間與 worker concurrency 標準庫適合解析「已被邊界控制」的 JSON；它不是替你完成所有資源治理的防火牆。

十二. 實戰：安全轉換訂單 JSONL
#

把前面的策略合在一個可重跑流程裡：

from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
import json
def clean_order(row: dict[str, Any]) -> dict[str, Any]:
    required = {"id", "customer", "total"}
    missing = required - row.keys()
    if missing:
        raise ValueError(f"missing fields: {sorted(missing)}")
    try:
        total = Decimal(str(row["total"]))
    except InvalidOperation as error:
        raise ValueError("invalid total") from error
    if total < 0:
        raise ValueError("total must be non-negative")
    return {
        "id": str(row["id"]),
        "customer": str(row["customer"]).strip(),
        "total": format(total.quantize(Decimal("0.01")), "f"),
    }
def convert(source: Path, target: Path, rejected: Path) -> None:
    with target.open("w", encoding="utf-8") as good_file, \
         rejected.open("w", encoding="utf-8") as bad_file:
        for line_number, line in enumerate(
            source.open("r", encoding="utf-8"),
            start=1,
        ):
            try:
                value = json.loads(
                    line,
                    parse_float=Decimal,
                    parse_constant=reject_constant,
                    object_pairs_hook=unique_object,
                )
                if not isinstance(value, dict):
                    raise TypeError("object required")
                cleaned = clean_order(value)
                good_file.write(json.dumps(cleaned, ensure_ascii=False) + "\n")
            except (json.JSONDecodeError, TypeError, ValueError) as error:
                failure = {"line": line_number, "error": str(error)}
                bad_file.write(json.dumps(failure, ensure_ascii=False) + "\n")

這個範例刻意不把原始失敗資料寫進 rejected 檔，避免敏感內容被複製一份。正式環境還可以加入 temporary file + atomic replace，避免中途失敗留下看似完整的結果；流程可參考 tempfile 實戰。

十三. 怎麼選工具？
#

先從零依賴、hooks 完整的 json 開始通常很合理。當 profiling 證明序列化是瓶頸，再評估直接輸出 bytes 的 orjson；當資料契約與 typed decode 才是痛點，再看 msgspec。不要只因為 benchmark 很漂亮，就把整個專案的 wire format 換掉。