閉じる

PythonでWeb APIの複雑なJSONを安全にパースする方法

外部のWeb APIが返すJSONは、最初は簡単に見えても運用が進むと複雑になりがちです。

キーの欠損や型揺れ、スキーマ変更は避けられません。

本記事ではPythonで複雑なJSONを安全に取得・パースするための設計と実装を、例外処理・型ヒント・バリデーション・実践パターンまで段階的に解説します。

目次
  1. 複雑なJSONを扱う前提(Python×Web API)
  2. PythonでJSONを安全に取得・パースする基本
  3. 型で守るJSONパース設計(バリデーション)
  4. 複雑なJSON処理の実践パターン
  5. まとめ

複雑なJSONを扱う前提(Python×Web API)

Web APIのJSONが複雑になる理由

Web APIは機能追加やA/Bテストにより、レスポンスの項目が増減したり、意味を保ったまま表現が変わることがあります。

特に、配列のネストや異なる粒度のオブジェクトが混在すると、単純な辞書アクセスでは破綻します。

さらに、同じキーでも文脈により数値または文字列が返るなどの型揺れも発生しやすく、慎重なパースが求められます。

欠損キー(null)や型揺れのリスク

欠損はKeyErrorやTypeErrorを誘発し、型揺れは計算や比較でValueErrorを引き起こします。

例えば、価格が数値のときと文字列のときが混在すると、単純な算術は失敗します。

常に存在する前提でアクセスしない、安全なデフォルト値を設ける、型の正規化を行うことが重要です。

スキーマ変更を想定した設計方針

後方互換を意識して、次の方針を取り入れます。

余剰フィールドは無視、必須フィールドは明確化、曖昧な型はOptionalやUnionで表現します。

さらにバリデーション層を用意し、不正データは早期に弾くことで下流のロジックを単純化します。

下表は典型的な失敗と対策の対応です。

失敗の種類例検知と対策
ネットワーク/タイムアウト応答が遅いタイムアウト設定とリトライ、バックオフ
HTTPエラー429, 5xxraise_for_statusで検知、再試行ポリシ
JSON不正HTMLや空レスJSONDecodeError捕捉、ログ保存
欠損・型揺れpriceがnull/文字列getとデフォルト、isinstanceと正規化
スキーマ変更キー名変更/追加バリデータ層、extra="ignore"、エイリアス

PythonでJSONを安全に取得・パースする基本

requestsでWeb APIからJSON取得(タイムアウト/リトライ)

HTTPは失敗が前提です。

接続/読み取りタイムアウト、指数バックオフ付きリトライを標準実装にします。

Python
# python
# 安全にHTTP GETしJSONを取得するユーティリティ
from typing import Any, Dict, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def _make_session() -> requests.Session:
    """指数バックオフ付きのリトライを組み込んだセッションを生成します。"""
    session = requests.Session()
    retry = Retry(
        total=3,                # 総リトライ回数
        connect=3,              # 接続時のリトライ
        read=3,                 # 読み取り時のリトライ
        backoff_factor=0.5,     # 指数バックオフ係数(0.5, 1.0, 2.0秒...)
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"],  # 冪等性を確認のうえ設定
        raise_on_status=False
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

def fetch_json(url: str, params: Optional[Dict[str, Any]] = None,
               timeout: tuple[float, float] = (3.05, 10.0)) -> Dict[str, Any]:
    """
    Web APIからJSONを取得します。
    timeoutは(接続タイムアウト, 読み取りタイムアウト)を指定します。
    """
    session = _make_session()
    resp = session.get(url, params=params, timeout=timeout)
    resp.raise_for_status()  # HTTP 4xx/5xxを例外に
    return resp.json()       # JSONDecodeError(ValueError)が投げられる可能性あり

HTTPエラーとJSONDecodeErrorの例外処理

ネットワーク、HTTP、JSONのそれぞれを分けて扱うと解析が容易です。

ログにはHTTPステータスやレスポンスボディの一部を記録すると原因追跡に役立ちます。

Python
# python
# fetch_jsonを安全に呼び出し、各例外をハンドリングする例
from json import JSONDecodeError

def get_payload_or_none(url: str):
    try:
        return fetch_json(url)
    except requests.exceptions.Timeout:
        # タイムアウト(接続/読み取り)は再試行やフォールバック
        print("Timeout. Retrying later...")
    except requests.exceptions.HTTPError as e:
        # ステータスコードと一部本文
        print(f"HTTP error: {e.response.status_code} body={e.response.text[:200]!r}")
    except JSONDecodeError as e:
        # JSON不正(HTMLや空レスポンス)
        print(f"Invalid JSON: {e}")
    except requests.exceptions.RequestException as e:
        # その他のネットワークエラー
        print(f"Request failed: {e}")
    return None

dict/listの安全なアクセス(in/get/try)

辞書はinやgetで安全にアクセスします。

存在しない可能性がある場合はデフォルト値を使います。

Python
# python
# サンプルJSON(実際はAPIから取得する想定)
payload = {
    "user": {"id": 123, "name": "Alice"},
    "items": [
        {"sku": "A-1", "price": "1200", "tags": ["new", "sale"]},
        {"sku": "B-9", "price": None}
    ],
    "meta": {"next": None}
}

# 安全なアクセス例
user = payload.get("user", {})                    # userがなければ空dict
username = user.get("name", "unknown")            # nameがなければ"unknown"
has_tags_first = "tags" in payload.get("items", [{}])[0]  # inで存在判定

print(username, has_tags_first)
実行結果
Alice True

ネストしたキーの安全参照(デフォルト値と早期return)

深いネストは早期returnでガードを積み重ねると読みやすくなります。

Python
# python
from typing import Optional, Any, Dict

def next_cursor(data: Dict[str, Any]) -> Optional[str]:
    """レスポンスのmeta.nextを返す。なければNone。"""
    meta = data.get("meta")
    if not isinstance(meta, dict):
        return None
    nxt = meta.get("next")
    if nxt in (None, ""):
        return None
    if not isinstance(nxt, str):
        return None
    return nxt

print(next_cursor(payload))
実行結果
None

値の型チェックとNone対策(isinstance/Optional)

型揺れを正規化します。

例えばpriceが数値または文字列の場合に浮動小数へ寄せます。

Python
# python
from typing import Optional

def normalize_price(raw: object) -> Optional[float]:
    """数値/数値文字列をfloatへ、Noneや空文字はNoneへ正規化します。"""
    if raw in (None, ""):
        return None
    if isinstance(raw, (int, float)):
        return float(raw)
    if isinstance(raw, str):
        try:
            return float(raw.replace(",", ""))  # "1,200"にも対応
        except ValueError:
            return None
    return None

prices = [normalize_price(it.get("price")) for it in payload.get("items", [])]
print(prices)
実行結果
[1200.0, None]

補足として、デバッグ時はjson.dumpsで整形表示すると差分が見やすくなります。

Python
# python
import json
print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
実行結果
{
  "items": [
    {
      "price": "1200",
      "sku": "A-1",
      "tags": [
        "new",
        "sale"
      ]
    },
    {
      "price": null,
      "sku": "B-9"
    }
  ],
  "meta": {
    "next": null
  },
  "user": {
    "id": 123,
    "name": "Alice"
  }
}

型で守るJSONパース設計(バリデーション)

TypedDict(NotRequired)でJSONスキーマを表現

TypedDictは辞書ベースのJSONに型を与え、エディタ補完や静的解析の助けになります。

NotRequiredで任意フィールドを表現できます。

Python
# python
# Python 3.11未満はtyping_extensionsからNotRequiredを使う
try:
    from typing import TypedDict, NotRequired
except ImportError:
    from typing_extensions import TypedDict, NotRequired

from typing import Optional, List

class ItemTD(TypedDict):
    sku: str
    price: NotRequired[Optional[float]]   # 任意かつNone許容
    tags: NotRequired[List[str]]          # 任意

class PayloadTD(TypedDict):
    user: NotRequired[dict]
    items: NotRequired[List[ItemTD]]
    meta: NotRequired[dict]

静的型付けは実行時検証はしませんが、コードの意図を明文化し、レビュや保守で威力を発揮します。

dataclassでJSON→オブジェクトをマッピング

dataclassesは軽量で依存が不要です。

コンバータを自前実装し、入力の曖昧さを吸収します。

Python
# python
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any

@dataclass
class Item:
    sku: str
    price: Optional[float] = None
    tags: List[str] = field(default_factory=list)

    @staticmethod
    def from_dict(d: Dict[str, Any]) -> "Item":
        """生JSONから安全にItemへ変換します。"""
        sku = str(d.get("sku", "")).strip()
        price = normalize_price(d.get("price"))
        tags = d.get("tags") or []
        if not isinstance(tags, list):
            tags = []
        tags = [str(t) for t in tags]  # 予防的に文字列化
        return Item(sku=sku, price=price, tags=tags)

items = [Item.from_dict(d) for d in payload.get("items", [])]
print(items[0].sku, items[0].price, items[0].tags)
実行結果
A-1 1200.0 ['new', 'sale']

pydanticで厳密なバリデーションと型変換

実行時バリデーションと型変換を一括で行うならpydanticが強力です。

必須/任意、余剰フィールド、型変換を宣言的に管理できます。

Python
# python
# pydantic v2想定
from datetime import datetime
from typing import Optional, List, Union
from pydantic import BaseModel, Field, ValidationError, field_validator, ConfigDict

class ItemModel(BaseModel):
    model_config = ConfigDict(extra="ignore")  # 想定外のフィールドは無視(後方互換)
    sku: str
    price: Optional[float] = None             # 数値や数値文字列を自動でfloatへ
    tags: List[str] = Field(default_factory=list)
    created_at: Optional[datetime] = None     # 文字列→datetimeを自動パース

    @field_validator("price", mode="before")
    @classmethod
    def parse_price(cls, v):
        # ""やNoneはNoneへ、それ以外はpydanticに委譲
        if v in (None, ""):
            return None
        if isinstance(v, str):
            v = v.replace(",", "")
        return v

# キー名が将来変わることへの対策(エイリアス)
class ItemModelV2(ItemModel):
    # 新APIでは"code"に改名されたと仮定
    sku: str = Field(alias="sku", validation_alias="code|sku")

try:
    m = ItemModel(sku="A-1", price="1,200", tags=["sale"], created_at="2024-07-01T12:34:56Z")
    print(m.model_dump())
except ValidationError as e:
    print(e)

try:
    # "code"のみを持つ新スキーマでも受け入れる
    mv2 = ItemModelV2(code="B-9", price=None, tags=["new"])
    print(mv2.sku, mv2.price, mv2.tags)
except ValidationError as e:
    print(e)
実行結果
{'sku': 'A-1', 'price': 1200.0, 'tags': ['sale'], 'created_at': datetime.datetime(2024, 7, 1, 12, 34, 56, tzinfo=TzInfo(UTC))}
B-9 None ['new']

Optional/Unionで不安定なフィールドに対応

フィールドがintまたはstrで来る場合、Union[int, str]で受け、必要に応じて正規化します。

pydanticはUnionの順番に従って解決します。

Python
# python
from typing import Union

class MaybeIntModel(BaseModel):
    value: Optional[Union[int, str]] = None

    @property
    def value_as_int(self) -> Optional[int]:
        if self.value is None:
            return None
        if isinstance(self.value, int):
            return self.value
        try:
            return int(self.value)
        except ValueError:
            return None

print(MaybeIntModel(value="42").value_as_int, MaybeIntModel(value="x").value_as_int)
実行結果
42 None

複雑なJSON処理の実践パターン

ネスト配列の走査とフィルタ(内包表記)

ネストした配列から条件で抽出します。

Noneを除外しながらの変換がポイントです。

Python
# python
# "sale"タグが付いた商品SKUを抽出し、価格が1000以上のものに限定
def sale_skus_over(items_raw, min_price: float = 1000.0):
    return [
        it.get("sku")
        for it in items_raw
        if isinstance(it, dict)
        and "sale" in (it.get("tags") or [])
        and (p := normalize_price(it.get("price"))) is not None
        and p >= min_price
    ]

print(sale_skus_over(payload["items"]))
実行結果
['A-1']

安全な抽出ヘルパーを作る(safe_get)

ドットとインデックスでネストを横断する小さなヘルパーを用意すると、記述が簡潔になります。

Python
# python
from typing import Any, Iterable

def safe_get(data: Any, path: str, default: Any = None) -> Any:
    """
    path例: "items[0].tags[1]" や "meta.next"
    存在しなければdefaultを返します。
    """
    cur = data
    for part in path.split("."):
        if "[" in part:
            # キーとインデックスを順に処理
            key, rest = part.split("[", 1)
            if key:
                if not isinstance(cur, dict) or key not in cur:
                    return default
                cur = cur[key]
            while rest:
                idx_str, rest = rest.split("]", 1)
                idx = int(idx_str)
                if not isinstance(cur, list) or idx >= len(cur):
                    return default
                cur = cur[idx]
                if rest.startswith("["):
                    rest = rest[1:]
                elif rest.startswith(".") or not rest:
                    break
        else:
            if not isinstance(cur, dict) or part not in cur:
                return default
            cur = cur[part]
    return cur

print(safe_get(payload, "items[0].tags[1]", default="N/A"))
print(safe_get(payload, "meta.next", default=None))
実行結果
sale
None

部分的な正規化とフラット化(必要な項目だけ)

分析や保存のために、必要最小限の項目を抽出してフラット化すると扱いやすくなります。

Python
# python
def flatten_items(data: dict) -> list[dict]:
    """itemsだけを正規化してフラットなレコードへ"""
    out: list[dict] = []
    for it in data.get("items", []):
        if not isinstance(it, dict):
            continue
        record = {
            "sku": str(it.get("sku", "")),
            "price": normalize_price(it.get("price")),
            "has_sale": "sale" in (it.get("tags") or []),
        }
        out.append(record)
    return out

flat = flatten_items(payload)
print(flat)
実行結果
[{'sku': 'A-1', 'price': 1200.0, 'has_sale': True}, {'sku': 'B-9', 'price': None, 'has_sale': False}]

フィールド追加/変更への耐性(後方互換の設計)

変更に強くするコツは次の通りです。

余剰フィールドは無視、未知フィールドはログに記録、重要キーはエイリアスで受ける、旧新どちらの表現も解釈できる正規化を実装します。

pydanticならextra="ignore"、エイリアス(validation_alias)を活用します。

生の辞書処理でも存在チェックとデフォルト値を徹底します。

Python
# python
# 旧: "price"  新: "unit_price" を両方受ける例
def pick_price(d: dict) -> Optional[float]:
    for key in ("unit_price", "price"):
        if key in d:
            return normalize_price(d.get(key))
    return None

print(pick_price({"unit_price": "980"}), pick_price({"price": 1000}), pick_price({}))
実行結果
980.0 1000.0 None

ログとテストデータで再現性を確保

バグ再現には生レスポンスのスナップショットが不可欠です。

PIIに配慮しつつ、json.dumps(..., indent=2)で整形し、IDやトークンをマスキングして保存します。

ユニットテストではこの固定JSONを使って回帰を防止します。

Python
# python
import json
from pathlib import Path

def save_response_snapshot(name: str, data: dict) -> Path:
    """レスポンスをスナップショット保存(機密値は呼び出し側でマスク)。"""
    p = Path("snapshots") / f"{name}.json"
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    return p

path = save_response_snapshot("example", payload)
print(f"saved: {path}")
実行結果
saved: snapshots/example.json

まとめ

複雑なJSONを安全に扱うには、取得の堅牢化(タイムアウト/リトライ)、例外の正しい分離とログ、安全な辞書アクセスと早期return、型ヒントとバリデーションによる境界の明確化が重要です。

さらに、必要な項目だけを正規化してフラット化し、将来のスキーマ変更に備えた後方互換の設計を取り入れることで、運用コストを大幅に下げられます。

デバッグ時はjson.dumps(..., indent=2)でスナップショットを保存し、テストデータとして再利用してください。

バリデーション層で不正を早期に弾き、ドメイン層は綺麗なデータだけを扱うという境界づくりが、最終的な安定性と開発速度を両立させます。

Python 実践TIPS - データ操作
この記事を書いた人
エーテリア編集部
エーテリア編集部

人気のPythonを初めて学ぶ方向けに、文法の基本から小さな自動化まで、実際に手を動かして理解できる記事を書いています。

クラウドSSLサイトシールは安心の証です。

URLをコピーしました！