Python JSON完全マスター【データ変換・API連携を徹底攻略】
JSON(JavaScript Object Notation)は、軽量なデータ交換フォーマットとして、Web API、設定ファイル、データストレージなど幅広い用途で使用されています。本記事では、PythonでのJSON操作を基本から応用まで、実用的なサンプルコードとともに徹底解説します。
JSONとは
JSONは、人間が読みやすく、機械が解析しやすいテキストベースのデータ形式です。主な特徴:
- 軽量: XMLより簡潔
- 可読性: 人間が読みやすい構造
- 言語非依存: 多くのプログラミング言語でサポート
- Web標準: REST APIの標準データ形式
- 構造化データ: ネストした複雑なデータを表現可能
JSONの基本構文
{
"name": "田中太郎",
"age": 30,
"is_active": true,
"skills": ["Python", "JavaScript", "SQL"],
"address": {
"prefecture": "東京都",
"city": "渋谷区"
},
"score": null
}
Pythonの標準jsonモジュール
基本的なインポート
import json
PythonのjsonモジュールはPython標準ライブラリに含まれており、追加インストールは不要です。
JSONデータの読み込み(デシリアライゼーション)
json.loads() – 文字列からPythonオブジェクトへ
import json
json_string = '{"name": "田中", "age": 30, "city": "東京"}'
data = json.loads(json_string)
print(data["name"]) # 田中
print(type(data)) # <class 'dict'>
json.load() – ファイルから読み込み
import json
with open('data.json', 'r', encoding='utf-8') as file:
data = json.load(file)
print(data)
複雑なJSONデータの処理
import json
json_data = '''
{
"users": [
{"id": 1, "name": "Alice", "scores": [85, 90, 78]},
{"id": 2, "name": "Bob", "scores": [92, 88, 95]}
],
"total": 2
}
'''
data = json.loads(json_data)
for user in data["users"]:
average = sum(user["scores"]) / len(user["scores"])
print(f'{user["name"]}: {average:.1f}')
JSONデータの書き出し(シリアライゼーション)
json.dumps() – Pythonオブジェクトから文字列へ
import json
data = {
"name": "山田花子",
"age": 25,
"hobbies": ["読書", "映画鑑賞"],
"married": False
}
json_string = json.dumps(data, ensure_ascii=False, indent=2)
print(json_string)
json.dump() – ファイルに書き出し
import json
data = {"temperature": 25.5, "humidity": 60, "location": "Tokyo"}
with open('weather.json', 'w', encoding='utf-8') as file:
json.dump(data, file, ensure_ascii=False, indent=2)
美しいJSON出力の設定
import json
data = {"items": [{"name": "商品A", "price": 1000}, {"name": "商品B", "price": 2000}]}
# 整形されたJSON出力
formatted_json = json.dumps(
data,
ensure_ascii=False, # 日本語をそのまま出力
indent=2, # インデント
separators=(',', ': ') # セパレータ
)
print(formatted_json)
データ型の対応関係
PythonとJSONの型変換
| Python型 | JSON型 | 変換例 |
|---|---|---|
| dict | object | {"key": "value"} |
| list, tuple | array | [1, 2, 3] |
| str | string | "text" |
| int, float | number | 42, 3.14 |
| True, False | true, false | true |
| None | null | null |
import json
python_data = {
"text": "Hello",
"number": 42,
"float": 3.14,
"boolean": True,
"null_value": None,
"list": [1, 2, 3],
"tuple": (4, 5, 6) # リストとして変換される
}
json_string = json.dumps(python_data)
print(json_string)
エラーハンドリング
JSONDecodeErrorの処理
import json
def safe_json_parse(json_string):
try:
return json.loads(json_string)
except json.JSONDecodeError as e:
print(f"JSONパースエラー: {e}")
return None
# 正常なJSON
result1 = safe_json_parse('{"name": "test"}')
print(result1)
# 不正なJSON
result2 = safe_json_parse('{"name": test}') # クォートなし
print(result2) # None
ファイル読み込み時のエラー処理
import json
def load_json_file(filename):
try:
with open(filename, 'r', encoding='utf-8') as file:
return json.load(file)
except FileNotFoundError:
print(f"ファイルが見つかりません: {filename}")
except json.JSONDecodeError as e:
print(f"JSONフォーマットエラー: {e}")
except Exception as e:
print(f"予期しないエラー: {e}")
return {}
data = load_json_file('config.json')
実用的な応用例
Web APIからのデータ取得
import requests
import json
def fetch_api_data(url):
response = requests.get(url)
if response.status_code == 200:
return response.json() # 自動的にJSONをパース
return None
# GitHub APIの例
user_data = fetch_api_data('https://api.github.com/users/octocat')
if user_data:
print(f"ユーザー: {user_data['name']}")
print(f"リポジトリ数: {user_data['public_repos']}")
設定ファイルの管理
import json
import os
class ConfigManager:
def __init__(self, config_path='config.json'):
self.config_path = config_path
self.config = self.load_config()
def load_config(self):
if os.path.exists(self.config_path):
with open(self.config_path, 'r', encoding='utf-8') as file:
return json.load(file)
return {"database": {"host": "localhost", "port": 5432}}
def save_config(self):
with open(self.config_path, 'w', encoding='utf-8') as file:
json.dump(self.config, file, ensure_ascii=False, indent=2)
def get(self, key, default=None):
return self.config.get(key, default)
# 使用例
config = ConfigManager()
db_host = config.get('database', {}).get('host', 'localhost')
print(f"データベースホスト: {db_host}")
ログデータの処理
import json
from datetime import datetime
def create_log_entry(level, message, **kwargs):
log_entry = {
"timestamp": datetime.now().isoformat(),
"level": level,
"message": message,
**kwargs
}
return json.dumps(log_entry, ensure_ascii=False)
def parse_log_file(filename):
logs = []
with open(filename, 'r', encoding='utf-8') as file:
for line in file:
try:
log_data = json.loads(line.strip())
logs.append(log_data)
except json.JSONDecodeError:
continue
return logs
# ログ出力例
log = create_log_entry("ERROR", "データベース接続失敗", user_id=123)
print(log)
カスタムエンコーダーとデコーダー
日付時刻の処理
import json
from datetime import datetime, date
class DateTimeEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (datetime, date)):
return obj.isoformat()
return super().default(obj)
# 使用例
data = {
"name": "イベント",
"created_at": datetime.now(),
"event_date": date.today()
}
json_string = json.dumps(data, cls=DateTimeEncoder, ensure_ascii=False)
print(json_string)
カスタムオブジェクトのシリアライゼーション
import json
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def to_dict(self):
return {"name": self.name, "age": self.age}
@classmethod
def from_dict(cls, data):
return cls(data["name"], data["age"])
def person_encoder(obj):
if isinstance(obj, Person):
return obj.to_dict()
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
# シリアライゼーション
person = Person("田中", 30)
json_string = json.dumps(person, default=person_encoder, ensure_ascii=False)
print(json_string)
# デシリアライゼーション
data = json.loads(json_string)
restored_person = Person.from_dict(data)
print(f"復元: {restored_person.name}, {restored_person.age}")
JSONスキーマ検証
基本的なバリデーション
import json
def validate_user_data(data):
required_fields = ["name", "email", "age"]
if not isinstance(data, dict):
return False, "データは辞書である必要があります"
for field in required_fields:
if field not in data:
return False, f"必須フィールド '{field}' がありません"
if not isinstance(data["age"], int) or data["age"] < 0:
return False, "年齢は0以上の整数である必要があります"
return True, "バリデーション成功"
# テスト
test_data = {"name": "太郎", "email": "taro@example.com", "age": 25}
is_valid, message = validate_user_data(test_data)
print(f"バリデーション結果: {message}")
JSONSchemaライブラリを使用した検証
# pip install jsonschema が必要
import json
import jsonschema
schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer", "minimum": 0},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "age", "email"]
}
def validate_with_schema(data, schema):
try:
jsonschema.validate(data, schema)
return True, "バリデーション成功"
except jsonschema.ValidationError as e:
return False, str(e)
# テスト
test_data = {"name": "花子", "age": 30, "email": "hanako@example.com"}
is_valid, message = validate_with_schema(test_data, schema)
print(f"スキーマ検証: {message}")
パフォーマンス最適化
大きなJSONファイルの効率的な処理
import json
def process_large_json_streaming(filename):
"""大きなJSONファイルを効率的に処理"""
with open(filename, 'r', encoding='utf-8') as file:
# JSONファイルが配列の場合の処理例
data = json.load(file)
if isinstance(data, list):
for i, item in enumerate(data):
# バッチ処理
if i % 1000 == 0:
print(f"処理中: {i}件目")
yield item
# 使用例
# for item in process_large_json_streaming('large_data.json'):
# # 各アイテムを処理
# pass
メモリ効率的なJSON処理
import json
import ijson # pip install ijson が必要
def parse_large_json_efficiently(filename):
"""大容量JSONファイルの効率的な解析"""
items = []
with open(filename, 'rb') as file:
parser = ijson.parse(file)
for prefix, event, value in parser:
if prefix.endswith('.name') and event == 'string':
items.append(value)
if len(items) >= 100: # バッチサイズ
yield items
items = []
if items:
yield items
# 通常のjsonライブラリでの簡易版
def simple_batch_processor(data, batch_size=100):
for i in range(0, len(data), batch_size):
yield data[i:i + batch_size]
JSONとCSVの相互変換
JSONからCSVへの変換
import json
import csv
def json_to_csv(json_file, csv_file):
with open(json_file, 'r', encoding='utf-8') as jf:
data = json.load(jf)
if not data:
return
with open(csv_file, 'w', newline='', encoding='utf-8') as cf:
writer = csv.DictWriter(cf, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
# 使用例
sample_data = [
{"name": "Alice", "age": 30, "city": "Tokyo"},
{"name": "Bob", "age": 25, "city": "Osaka"}
]
with open('sample.json', 'w', encoding='utf-8') as f:
json.dump(sample_data, f, ensure_ascii=False, indent=2)
json_to_csv('sample.json', 'output.csv')
CSVからJSONへの変換
import json
import csv
def csv_to_json(csv_file, json_file):
data = []
with open(csv_file, 'r', encoding='utf-8') as cf:
reader = csv.DictReader(cf)
for row in reader:
data.append(row)
with open(json_file, 'w', encoding='utf-8') as jf:
json.dump(data, jf, ensure_ascii=False, indent=2)
# csv_to_json('input.csv', 'output.json')
Web開発での活用
FlaskでのJSON API
from flask import Flask, jsonify, request
import json
app = Flask(__name__)
@app.route('/api/users', methods=['GET'])
def get_users():
users = [
{"id": 1, "name": "太郎", "email": "taro@example.com"},
{"id": 2, "name": "花子", "email": "hanako@example.com"}
]
return jsonify(users)
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
if not data or 'name' not in data:
return jsonify({"error": "名前は必須です"}), 400
new_user = {
"id": 3,
"name": data["name"],
"email": data.get("email", "")
}
return jsonify(new_user), 201
# if __name__ == '__main__':
# app.run(debug=True)
FastAPIでのJSON処理
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import json
app = FastAPI()
class User(BaseModel):
name: str
age: int
email: str
@app.post("/users/")
async def create_user(user: User):
# Pydanticモデルは自動的にJSONに変換される
return {"message": "ユーザー作成成功", "user": user}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# ダミーデータ
if user_id == 1:
return {"id": 1, "name": "テストユーザー", "age": 30}
raise HTTPException(status_code=404, detail="ユーザーが見つかりません")
デバッグとトラブルシューティング
JSON構文エラーの特定
import json
import re
def debug_json_error(json_string):
try:
json.loads(json_string)
print("JSON形式は正しいです")
except json.JSONDecodeError as e:
print(f"エラー位置: {e.pos}")
print(f"エラー行: {e.lineno}")
print(f"エラー列: {e.colno}")
print(f"エラー内容: {e.msg}")
# エラー箇所を表示
lines = json_string.split('\n')
if e.lineno <= len(lines):
error_line = lines[e.lineno - 1]
print(f"問題のある行: {error_line}")
print(f"{'':>{e.colno-1}}^")
# テスト
bad_json = '''
{
"name": "test",
"age": 30,
"city": "Tokyo" // コメントは無効
}
'''
debug_json_error(bad_json)
JSONの差分比較
import json
def compare_json(json1, json2):
def find_differences(obj1, obj2, path=""):
differences = []
if type(obj1) != type(obj2):
differences.append(f"{path}: 型が異なります ({type(obj1)} vs {type(obj2)})")
return differences
if isinstance(obj1, dict):
all_keys = set(obj1.keys()) | set(obj2.keys())
for key in all_keys:
new_path = f"{path}.{key}" if path else key
if key not in obj1:
differences.append(f"{new_path}: 左側にキーがありません")
elif key not in obj2:
differences.append(f"{new_path}: 右側にキーがありません")
else:
differences.extend(find_differences(obj1[key], obj2[key], new_path))
elif isinstance(obj1, list):
if len(obj1) != len(obj2):
differences.append(f"{path}: 配列長が異なります ({len(obj1)} vs {len(obj2)})")
for i in range(min(len(obj1), len(obj2))):
differences.extend(find_differences(obj1[i], obj2[i], f"{path}[{i}]"))
elif obj1 != obj2:
differences.append(f"{path}: 値が異なります ({obj1} vs {obj2})")
return differences
return find_differences(json1, json2)
# テスト
json_a = {"name": "Alice", "age": 30, "skills": ["Python", "JavaScript"]}
json_b = {"name": "Alice", "age": 31, "skills": ["Python", "Java"]}
differences = compare_json(json_a, json_b)
for diff in differences:
print(diff)
JSONの圧縮と最適化
不要な空白の除去
import json
def minimize_json(json_string):
"""JSONから不要な空白を除去"""
data = json.loads(json_string)
return json.dumps(data, separators=(',', ':'), ensure_ascii=False)
# テスト
formatted_json = '''
{
"name": "テスト",
"data": [
1,
2,
3
]
}
'''
minimized = minimize_json(formatted_json)
print(f"元のサイズ: {len(formatted_json)} bytes")
print(f"圧縮後: {len(minimized)} bytes")
print(f"圧縮率: {(1 - len(minimized)/len(formatted_json))*100:.1f}%")
まとめ
PythonでのJSON操作は、標準ライブラリのjsonモジュールにより簡単かつ効率的に実行できます。Web開発、データ分析、設定管理など様々な場面で活用される重要な技術です。
重要なポイント:
- json.loads/json.dumpsで基本的な変換をマスター
- エラーハンドリングで堅牢なコードを実装
- ensure_ascii=Falseで日本語を適切に処理
- カスタムエンコーダーで複雑なオブジェクトに対応
- パフォーマンスを考慮した大容量データ処理
本記事のサンプルコードを参考に、あなたのプロジェクトに最適なJSON処理を実装してください。適切なエラーハンドリングとパフォーマンス最適化により、実用的なシステムを構築できます。
参考文献
- Python公式jsonドキュメント
- JSON Schema公式サイト
- RFC 7159 – JSON仕様
- Flask/FastAPI公式ドキュメント
■プロンプトだけでオリジナルアプリを開発・公開してみた!!
■AI時代の第一歩!「AI駆動開発コース」はじめました!
テックジム東京本校で先行開始。
■テックジム東京本校
「武田塾」のプログラミング版といえば「テックジム」。
講義動画なし、教科書なし。「進捗管理とコーチング」で効率学習。
より早く、より安く、しかも対面型のプログラミングスクールです。
<短期講習>5日で5万円の「Pythonミニキャンプ」開催中。
<月1開催>放送作家による映像ディレクター養成講座
<オンライン無料>ゼロから始めるPython爆速講座

