PythonにおけるJSONデータ交換の実践ガイド

JSONは、Webアプリケーション間で構造化データを安全かつ効率的にやり取りするための標準フォーマットです。Pythonの組み込みjsonモジュールは、このフォーマットをネイティブにサポートし、開発者がシームレスにデータのシリアル化と逆シリアル化を行えるように設計されています。

基本的なマッピング規則

JSONの構文要素とPythonの対応型は以下の通りです:

JSON構文 Python型 変換例
オブジェクト({} dict {"title": "サンプル"}
配列([] list [1, 2, 3]
文字列 str "こんにちは"
数値(整数/浮動小数点) intfloat 423.14159
真偽値 bool trueTruefalseFalse
ヌル値 None nullNone

基本操作:4つの主要関数

データの双方向変換には、以下の関数ペアが中心となります:

import json

# Pythonオブジェクト → JSON文字列(エンコード)
payload = {"id": 789, "status": "active", "tags": ["dev", "api"]}
encoded = json.dumps(payload, ensure_ascii=False, indent=2)
print(encoded)
# 出力:
# {
#   "id": 789,
#   "status": "active",
#   "tags": [
#     "dev",
#     "api"
#   ]
# }

# JSON文字列 → Pythonオブジェクト(デコード)
decoded = json.loads(encoded)
assert decoded["id"] == 789

# Pythonオブジェクト → ファイル書き出し
with open("output.json", "w", encoding="utf-8") as f:
    json.dump(payload, f, ensure_ascii=False, indent=2)

# ファイル読み込み → Pythonオブジェクト
with open("output.json", "r", encoding="utf-8") as f:
    restored = json.load(f)

実用シナリオ

REST APIとの連携

外部サービスとの通信では、JSONペイロードの構築・解析が不可欠です:

import json
import requests

def send_api_request(endpoint: str, user_data: dict) -> dict | None:
    headers = {"Content-Type": "application/json"}
    
    try:
        # リクエストボディをJSONに変換
        json_payload = json.dumps(user_data, ensure_ascii=False)
        
        response = requests.post(
            endpoint,
            data=json_payload,
            headers=headers,
            timeout=5
        )
        response.raise_for_status()
        
        # 応答をJSONとしてパース
        return response.json()
        
    except requests.exceptions.RequestException as e:
        print(f"ネットワークエラー: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"応答のJSON解析失敗: {e}")
        return None

# 使用例
result = send_api_request(
    "https://httpbin.org/post",
    {"name": "山田太郎", "role": "developer"}
)
if result and "json" in result:
    print("受信データ:", result["json"])

設定管理の自動化

アプリケーション設定をJSONファイルで管理するクラスの簡易実装:

from pathlib import Path
import json

class ConfigManager:
    def __init__(self, config_path: str = "settings.json"):
        self.path = Path(config_path)
        self.data = self._load_config()
    
    def _load_config(self) -> dict:
        default = {
            "server": {"host": "127.0.0.1", "port": 8000},
            "logging": {"level": "INFO", "max_size_mb": 10}
        }
        
        if not self.path.exists():
            self.save(default)
            return default
        
        try:
            with open(self.path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, OSError):
            return default
    
    def save(self, data: dict):
        self.path.parent.mkdir(exist_ok=True)
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def get(self, key: str, default=None):
        keys = key.split(".")
        value = self.data
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return default
        return value

# 利用例
cfg = ConfigManager()
print("ポート番号:", cfg.get("server.port"))
cfg.save({"server": {"port": 8080}})

構造化データの永続化

ユーザーやセッション情報などの一時的データをJSONファイルに保存するユーティリティ:

import json
from datetime import datetime
from pathlib import Path

class DataStore:
    def __init__(self, base_dir: str = "storage"):
        self.base = Path(base_dir)
        self.base.mkdir(exist_ok=True)
    
    def store_record(self, category: str, identifier: str, content: dict):
        timestamp = datetime.now().isoformat()
        record = {
            "category": category,
            "identifier": identifier,
            "content": content,
            "stored_at": timestamp
        }
        
        filepath = self.base / f"{category}_{identifier}.json"
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(record, f, ensure_ascii=False, indent=2)
        return filepath
    
    def retrieve_record(self, category: str, identifier: str) -> dict | None:
        filepath = self.base / f"{category}_{identifier}.json"
        if not filepath.exists():
            return None
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, OSError):
            return None

# 使用例
store = DataStore()
store.store_record("user", "U123", {"name": "佐藤花子", "last_login": "2024-06-15"})
data = store.retrieve_record("user", "U123")
if data:
    print("取得したユーザー名:", data["content"]["name"])

高度なカスタマイズ

拡張型のシリアライズ対応

標準JSONでは表現できない型(日付、UUIDなど)を扱うカスタムエンコーダー:

import json
from datetime import datetime, date
from uuid import UUID

class FlexibleEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, UUID):
            return str(obj)
        if hasattr(obj, "__dict__"):
            return obj.__dict__
        return super().default(obj)

# シリアル化例
sample = {
    "created": datetime(2024, 6, 15, 14, 30),
    "session_id": UUID("a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8"),
    "metadata": {"version": "2.1.0"}
}

json_text = json.dumps(sample, cls=FlexibleEncoder, ensure_ascii=False)
print(json_text)

パフォーマンス最適化手法

大規模データ処理時の効率化テクニック:

  • コンパクト出力:`separators=(',', ':')`で空白削減(約20%サイズ削減)
  • 高速ライブラリ導入:`ujson`や`orjson`への置き換え(2〜5倍高速化)
  • ストリーミング解析:`json.JSONDecoder.raw_decode()`による逐次処理

重要な制限事項

  • 循環参照を持つオブジェクトはシリアライズ不可
  • バイナリデータ、複素数、セット(set)などは直接サポートされない
  • 信頼できないソースからのJSONは、悪意あるオブジェクトフックを介して脆弱性を引き起こす可能性がある
  • 非常に大きなJSONファイルはメモリ使用量を急増させる

タグ: Python JSON web-api Serialization configuration-management

6月14日 23:33 投稿