orjson

シリアライゼーションPythonRust高性能dataclassesdatetimeNumPy

ライブラリ

orjson

概要

orjsonは、Rustで書かれた超高速PythonのJSONライブラリです。標準jsonライブラリより6倍以上高速で、dataclassesやdatetimeなどのPython固有型をネイティブサポートします。2025年Python JSONライブラリの最有力候補として、パフォーマンス重視のアプリケーションでujsonからの移行が推奨されており、高速化が求められるAPI開発で急速に採用が増加しています。

詳細

orjson 3.10は、Rustの持つメモリ安全性と高速性をPythonエコシステムに持ち込む画期的なライブラリです。標準ライブラリと完全互換性を保ちながら、dataclasses、datetime、NumPy配列、UUID等のPython固有型をネイティブでサポートし、オプションによる細かな出力制御が可能です。RFC 3339準拠のdatetimeシリアライゼーション、microsecond精度の時刻処理、セキュアなUTF-8検証など、本格的なプロダクション環境での要求に応える高品質な実装を提供します。

主な特徴

超高速: 標準jsonライブラリより6倍以上高速
Python型対応: dataclasses、datetime、NumPyのネイティブサポート
RFC準拠: RFC 3339準拠のdatetimeシリアライゼーション
メモリ効率: Rustベースによる低メモリフットプリント
型安全: Rust由来の堅牢なメモリ安全性
互換性: 標準jsonライブラリとの完全互換API

メリット・デメリット

メリット

標準jsonライブラリと比較して圧倒的な高速性（6倍以上）
dataclasses、datetime、NumPy配列などPython固有型の標準サポート
RFC 3339準拠による国際標準に沿ったdatetime処理
Rustベースによる堅牢なメモリ安全性とクラッシュ耐性
標準ライブラリと同一のAPIによる簡単な置き換え
プロダクション環境での豊富な運用実績

デメリット

インストール時にコンパイル済みバイナリが必要（一部プラットフォームで制限）
標準ライブラリより大きなメモリフットプリント（バイナリサイズ）
カスタムエンコーダーの拡張性が標準ライブラリより限定的
デバッグ情報が限定的でトラブルシューティングが困難な場合がある
Rustツールチェーンの依存（ソースからビルドする場合）
小規模なデータでは標準ライブラリとの性能差が小さい

参考ページ

書き方の例

基本的なセットアップ

# orjsonのインストール
pip install orjson

# 依存関係と一緒にインストール
pip install "orjson>=3.10,<4"

# pipファイル（pipenv）での指定
echo 'orjson = "^3.10"' >> Pipfile

# requirements.txtでの指定
echo 'orjson >= 3.10,<4' >> requirements.txt

基本的なシリアライゼーション

import orjson
import datetime
import uuid

# 基本的なデータ構造
user_data = {
    "id": 123,
    "name": "田中太郎",
    "email": "[email protected]",
    "is_active": True,
    "balance": 1234.56,
    "tags": ["admin", "premium"],
    "metadata": {
        "created_at": datetime.datetime.now(),
        "last_login": None
    }
}

# シリアライゼーション
json_bytes = orjson.dumps(user_data)
print(f"JSON bytes: {json_bytes}")
print(f"Size: {len(json_bytes)} bytes")

# デシリアライゼーション
decoded_data = orjson.loads(json_bytes)
print(f"Decoded: {decoded_data}")

# 文字列からの読み込み
json_string = '{"name": "田中太郎", "age": 30, "active": true}'
parsed_data = orjson.loads(json_string)
print(f"Parsed: {parsed_data}")

# bytes、bytearray、memoryviewもサポート
data_from_bytes = orjson.loads(b'{"key": "value"}')
data_from_bytearray = orjson.loads(bytearray(b'{"key": "value"}'))
data_from_memoryview = orjson.loads(memoryview(b'{"key": "value"}'))

print("All input types supported:", data_from_bytes, data_from_bytearray, data_from_memoryview)

dataclassesとdatetimeサポート

import orjson
import dataclasses
import datetime
import zoneinfo
from typing import List, Optional

@dataclasses.dataclass
class User:
    id: int
    name: str
    email: str
    created_at: datetime.datetime
    last_login: Optional[datetime.datetime] = None
    is_active: bool = dataclasses.field(default=True)
    roles: List[str] = dataclasses.field(default_factory=list)

@dataclasses.dataclass  
class Organization:
    id: int
    name: str
    users: List[User]
    founded_date: datetime.date
    
# dataclassインスタンスの作成
user1 = User(
    id=1,
    name="田中太郎",
    email="[email protected]",
    created_at=datetime.datetime(2023, 1, 15, 10, 30, 0, tzinfo=zoneinfo.ZoneInfo("Asia/Tokyo")),
    last_login=datetime.datetime.now(zoneinfo.ZoneInfo("UTC")),
    roles=["admin", "user"]
)

user2 = User(
    id=2,
    name="佐藤花子", 
    email="[email protected]",
    created_at=datetime.datetime(2023, 6, 20, 14, 45, 30, 123456),
    roles=["user"]
)

org = Organization(
    id=100,
    name="株式会社サンプル",
    users=[user1, user2],
    founded_date=datetime.date(2020, 4, 1)
)

# dataclassesの自動シリアライゼーション
org_json = orjson.dumps(org)
print(f"Organization JSON: {org_json}")

# 各種datetime形式のサポート
datetime_examples = {
    "naive_datetime": datetime.datetime(2025, 1, 1, 12, 0, 0),
    "utc_datetime": datetime.datetime(2025, 1, 1, 12, 0, 0, tzinfo=zoneinfo.ZoneInfo("UTC")),
    "jst_datetime": datetime.datetime(2025, 1, 1, 12, 0, 0, tzinfo=zoneinfo.ZoneInfo("Asia/Tokyo")),
    "date_only": datetime.date(2025, 1, 1),
    "time_only": datetime.time(12, 30, 45, 123456)
}

# デフォルトの datetime シリアライゼーション
default_json = orjson.dumps(datetime_examples)
print(f"Default datetime: {default_json}")

# naive UTC として処理
naive_utc_json = orjson.dumps(datetime_examples, option=orjson.OPT_NAIVE_UTC)
print(f"Naive UTC: {naive_utc_json}")

# マイクロ秒を省略
no_microseconds_json = orjson.dumps(datetime_examples, option=orjson.OPT_OMIT_MICROSECONDS)
print(f"No microseconds: {no_microseconds_json}")

# UTC時刻にZ接尾辞を使用
utc_z_json = orjson.dumps(datetime_examples, option=orjson.OPT_UTC_Z)
print(f"UTC Z format: {utc_z_json}")

高度な設定とオプション

import orjson
import numpy as np
import decimal
import enum
from dataclasses import dataclass
from typing import Any

# カスタムenum定義
class Status(enum.Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    PENDING = "pending"

class Priority(enum.Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3

@dataclass
class Task:
    id: int
    title: str
    status: Status
    priority: Priority
    data: np.ndarray

# NumPy配列を含む複雑なデータ
task_data = Task(
    id=123,
    title="重要なタスク",
    status=Status.ACTIVE,
    priority=Priority.HIGH,
    data=np.array([[1, 2, 3], [4, 5, 6]])
)

# NumPyサポートを有効にしてシリアライゼーション
numpy_json = orjson.dumps(task_data, option=orjson.OPT_SERIALIZE_NUMPY)
print(f"NumPy serialization: {numpy_json}")

# 複数オプションの組み合わせ
combined_options = (
    orjson.OPT_NAIVE_UTC |           # naive datetimeをUTCとして処理
    orjson.OPT_SERIALIZE_NUMPY |     # NumPy配列をシリアライズ
    orjson.OPT_SORT_KEYS |           # キーをソート
    orjson.OPT_INDENT_2              # 2スペースでインデント
)

test_data = {
    "timestamp": datetime.datetime(2025, 1, 1, 12, 0, 0),
    "matrix": np.array([[1, 2], [3, 4]]),
    "z_key": "last",
    "a_key": "first",
    "m_key": "middle"
}

formatted_json = orjson.dumps(test_data, option=combined_options)
print(f"Combined options: {formatted_json.decode()}")

# 非文字列キーのサポート
non_str_keys_data = {
    uuid.uuid4(): "uuid key",
    datetime.datetime(2025, 1, 1): "datetime key",
    datetime.date(2025, 1, 1): "date key",
    42: "integer key"
}

non_str_json = orjson.dumps(non_str_keys_data, option=orjson.OPT_NON_STR_KEYS)
print(f"Non-string keys: {non_str_json}")

# 整数の厳密モード（53bit制限）
large_integers = {
    "safe_int": 9007199254740991,      # 2^53 - 1
    "unsafe_int": 9007199254740992     # 2^53
}

# 通常モード（64bit整数サポート）
normal_int_json = orjson.dumps(large_integers)
print(f"Normal integers: {normal_int_json}")

# 厳密モード（エラーが発生）
try:
    strict_int_json = orjson.dumps(large_integers, option=orjson.OPT_STRICT_INTEGER)
    print(f"Strict integers: {strict_int_json}")
except orjson.JSONEncodeError as e:
    print(f"Strict integer error: {e}")

カスタム型のシリアライゼーション

import orjson
import decimal
from datetime import datetime, timezone
from typing import Any

# カスタム型サポート用のdefault関数
def custom_default(obj: Any) -> Any:
    """カスタム型のシリアライゼーション処理"""
    
    if isinstance(obj, decimal.Decimal):
        # Decimal型を文字列として出力
        return str(obj)
    
    elif isinstance(obj, complex):
        # 複素数を辞書として出力
        return {"real": obj.real, "imag": obj.imag, "_type": "complex"}
    
    elif isinstance(obj, bytes):
        # バイト列をbase64エンコードして出力
        import base64
        return {"data": base64.b64encode(obj).decode(), "_type": "bytes"}
    
    elif isinstance(obj, set):
        # セットをリストとして出力
        return {"items": list(obj), "_type": "set"}
    
    elif hasattr(obj, '__dict__'):
        # カスタムクラスを辞書として出力
        return {**obj.__dict__, "_type": type(obj).__name__}
    
    # サポートしていない型はTypeErrorを発生
    raise TypeError(f"Type {type(obj)} is not JSON serializable")

# テスト用のカスタムクラス
class CustomObject:
    def __init__(self, name: str, value: int):
        self.name = name
        self.value = value
        self.created_at = datetime.now(timezone.utc)

# カスタム型を含むデータ
custom_data = {
    "decimal_value": decimal.Decimal("123.456789"),
    "complex_number": 3 + 4j,
    "byte_data": b"Hello, World!",
    "unique_items": {1, 2, 3, 4, 5},
    "custom_object": CustomObject("test", 42)
}

# カスタムシリアライゼーション
try:
    # default関数なしでは失敗
    orjson.dumps(custom_data)
except orjson.JSONEncodeError as e:
    print(f"Error without default: {e}")

# default関数ありで成功
custom_json = orjson.dumps(custom_data, default=custom_default, option=orjson.OPT_INDENT_2)
print(f"Custom serialization: {custom_json.decode()}")

# dataclassのpass-through例
@dataclass  
class SecureUser:
    id: int
    name: str
    password: str
    
def secure_default(obj: Any) -> Any:
    """セキュアなdataclassシリアライゼーション"""
    if isinstance(obj, SecureUser):
        # パスワードを除外してシリアライズ
        return {"id": obj.id, "name": obj.name}
    raise TypeError

secure_user = SecureUser(1, "田中太郎", "secret123")

# デフォルトのdataclassシリアライゼーション（パスワード含む）
default_user_json = orjson.dumps(secure_user)
print(f"Default dataclass: {default_user_json}")

# カスタムシリアライゼーション（パスワード除外）
secure_user_json = orjson.dumps(
    secure_user, 
    option=orjson.OPT_PASSTHROUGH_DATACLASS,
    default=secure_default
)
print(f"Secure dataclass: {secure_user_json}")

パフォーマンス最適化とベンチマーク

import orjson
import json
import time
import sys
from typing import List, Dict, Any
import datetime
import dataclasses

@dataclasses.dataclass
class PerformanceTestData:
    id: int
    name: str
    email: str
    created_at: datetime.datetime
    metadata: Dict[str, Any]
    tags: List[str]
    is_active: bool

def generate_test_data(count: int) -> List[PerformanceTestData]:
    """テスト用のデータ生成"""
    data = []
    for i in range(count):
        data.append(PerformanceTestData(
            id=i,
            name=f"User {i}",
            email=f"user{i}@example.com",
            created_at=datetime.datetime.now(),
            metadata={
                "department": f"Department {i % 10}",
                "salary": 50000 + (i * 100),
                "projects": [f"Project {j}" for j in range(i % 5)],
                "settings": {
                    "theme": "dark" if i % 2 else "light",
                    "notifications": i % 3 == 0
                }
            },
            tags=[f"tag{j}" for j in range(i % 3 + 1)],
            is_active=i % 10 != 0
        ))
    return data

def measure_time(func, *args, **kwargs):
    """実行時間測定"""
    start = time.perf_counter()
    result = func(*args, **kwargs)
    end = time.perf_counter()
    return result, (end - start) * 1000  # ミリ秒で返す

def benchmark_serialization(data_count: int = 10000):
    """シリアライゼーションのベンチマーク"""
    print(f"Benchmarking with {data_count} records...")
    
    # テストデータ生成
    test_data = generate_test_data(data_count)
    
    # orjson シリアライゼーション
    orjson_data, orjson_encode_time = measure_time(
        lambda: [orjson.dumps(item) for item in test_data]
    )
    
    # 標準json シリアライゼーション（dataclassは手動変換が必要）
    json_compatible_data = [dataclasses.asdict(item) for item in test_data]
    
    # datetimeを文字列に変換（標準jsonは直接処理できないため）
    def convert_datetime(obj):
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()
        return obj
    
    json_safe_data = []
    for item in json_compatible_data:
        item_copy = item.copy()
        item_copy['created_at'] = convert_datetime(item_copy['created_at'])
        json_safe_data.append(item_copy)
    
    json_data, json_encode_time = measure_time(
        lambda: [json.dumps(item) for item in json_safe_data]
    )
    
    # デシリアライゼーション
    orjson_decoded, orjson_decode_time = measure_time(
        lambda: [orjson.loads(data) for data in orjson_data]
    )
    
    json_decoded, json_decode_time = measure_time(
        lambda: [json.loads(data) for data in json_data]
    )
    
    # サイズ比較
    orjson_size = sum(len(data) for data in orjson_data)
    json_size = sum(len(data) for data in json_data)
    
    # 結果出力
    print("\n=== Serialization Performance ===")
    print(f"orjson encode: {orjson_encode_time:.2f}ms")
    print(f"json encode: {json_encode_time:.2f}ms")
    print(f"orjson speedup: {json_encode_time / orjson_encode_time:.1f}x")
    
    print("\n=== Deserialization Performance ===")
    print(f"orjson decode: {orjson_decode_time:.2f}ms")
    print(f"json decode: {json_decode_time:.2f}ms")
    print(f"orjson speedup: {json_decode_time / orjson_decode_time:.1f}x")
    
    print("\n=== Size Comparison ===")
    print(f"orjson size: {orjson_size / 1024:.2f} KB")
    print(f"json size: {json_size / 1024:.2f} KB")
    print(f"Size difference: {((json_size - orjson_size) / json_size * 100):.1f}%")
    
    print("\n=== Overall Performance ===")
    total_orjson_time = orjson_encode_time + orjson_decode_time
    total_json_time = json_encode_time + json_decode_time
    print(f"Total orjson time: {total_orjson_time:.2f}ms")
    print(f"Total json time: {total_json_time:.2f}ms")
    print(f"Overall speedup: {total_json_time / total_orjson_time:.1f}x")

# メモリ使用量の測定
def measure_memory_usage():
    """メモリ使用量の測定"""
    import tracemalloc
    
    # 大量データの生成
    large_data = generate_test_data(50000)
    
    # orjsonのメモリ使用量測定
    tracemalloc.start()
    orjson_result = [orjson.dumps(item) for item in large_data]
    orjson_current, orjson_peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    
    # 標準jsonのメモリ使用量測定
    json_safe_data = [dataclasses.asdict(item) for item in large_data]
    for item in json_safe_data:
        if 'created_at' in item:
            item['created_at'] = item['created_at'].isoformat()
    
    tracemalloc.start()
    json_result = [json.dumps(item) for item in json_safe_data]
    json_current, json_peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    
    print("\n=== Memory Usage ===")
    print(f"orjson peak memory: {orjson_peak / 1024 / 1024:.2f} MB")
    print(f"json peak memory: {json_peak / 1024 / 1024:.2f} MB")
    print(f"Memory efficiency: {json_peak / orjson_peak:.1f}x")

if __name__ == "__main__":
    # ベンチマーク実行
    benchmark_serialization(10000)
    measure_memory_usage()

エラーハンドリングとデバッグ

import orjson
import json
from typing import Any

def comprehensive_error_handling():
    """包括的なエラーハンドリングの例"""
    
    # 1. 基本的なエラーハンドリング
    invalid_data = {1, 2, 3}  # setは直接シリアライズできない
    
    try:
        result = orjson.dumps(invalid_data)
    except orjson.JSONEncodeError as e:
        print(f"Encode error: {e}")
        print(f"Error type: {type(e)}")
    
    # 2. 無効なUTF-8文字列の処理
    try:
        # 不正なサロゲートペア
        invalid_utf8 = '\ud800'  # 単独のハイサロゲート
        orjson.dumps(invalid_utf8)
    except orjson.JSONEncodeError as e:
        print(f"Invalid UTF-8 error: {e}")
        
        # 標準jsonとの比較
        print(f"Standard json handles it: {json.dumps(invalid_utf8)}")
    
    # 3. デコードエラーの処理
    try:
        # 無効なJSON
        invalid_json = b'{"key": value}'  # valueがクォートされていない
        orjson.loads(invalid_json)
    except orjson.JSONDecodeError as e:
        print(f"Decode error: {e}")
        print(f"Error position: line {getattr(e, 'lineno', 'unknown')}, column {getattr(e, 'colno', 'unknown')}")
    
    # 4. 無効なUTF-8バイト列の処理
    try:
        invalid_bytes = b'"\xed\xa0\x80"'  # 無効なUTF-8バイト列
        orjson.loads(invalid_bytes)
    except orjson.JSONDecodeError as e:
        print(f"Invalid UTF-8 bytes error: {e}")
        
        # 回避策: エラー処理付きでデコード
        try:
            recovered = orjson.loads(invalid_bytes.decode("utf-8", "replace"))
            print(f"Recovered data: {recovered}")
        except Exception as recovery_error:
            print(f"Recovery failed: {recovery_error}")
    
    # 5. 大きな整数の処理
    large_numbers = {
        "max_safe": 9007199254740991,     # 2^53 - 1
        "too_large": 9007199254740992,    # 2^53
        "way_too_large": 2**60
    }
    
    # 通常モード
    normal_result = orjson.dumps(large_numbers)
    print(f"Normal mode: {normal_result}")
    
    # 厳密モード
    try:
        strict_result = orjson.dumps(large_numbers, option=orjson.OPT_STRICT_INTEGER)
    except orjson.JSONEncodeError as e:
        print(f"Strict integer mode error: {e}")
    
    # 6. 循環参照の検出
    circular_data = {"name": "parent"}
    circular_data["self"] = circular_data
    
    try:
        orjson.dumps(circular_data)
    except orjson.JSONEncodeError as e:
        print(f"Circular reference error: {e}")
    
    # 7. カスタムdefault関数でのエラー処理
    def failing_default(obj: Any) -> Any:
        if isinstance(obj, set):
            return list(obj)
        # 明示的なTypeErrorを発生させる
        raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
    
    def safe_default(obj: Any) -> Any:
        """安全なdefault関数"""
        try:
            if isinstance(obj, set):
                return {"_type": "set", "items": list(obj)}
            elif hasattr(obj, '__dict__'):
                return {"_type": type(obj).__name__, "data": obj.__dict__}
            else:
                # フォールバック: 文字列表現
                return {"_type": "unknown", "str": str(obj)}
        except Exception as e:
            # 最終的なフォールバック
            return {"_type": "error", "message": str(e)}
    
    problematic_data = {
        "set_data": {1, 2, 3},
        "custom_object": object(),
        "lambda_func": lambda x: x
    }
    
    # 失敗するdefault関数
    try:
        orjson.dumps(problematic_data, default=failing_default)
    except orjson.JSONEncodeError as e:
        print(f"Default function error: {e}")
    
    # 安全なdefault関数
    safe_result = orjson.dumps(problematic_data, default=safe_default, option=orjson.OPT_INDENT_2)
    print(f"Safe default result: {safe_result.decode()}")

def debug_helper_functions():
    """デバッグ用のヘルパー関数"""
    
    def safe_dumps(obj: Any, **kwargs) -> bytes:
        """安全なorjson.dumps"""
        try:
            return orjson.dumps(obj, **kwargs)
        except orjson.JSONEncodeError as e:
            print(f"Serialization failed: {e}")
            # フォールバック: 標準jsonを使用
            try:
                fallback_result = json.dumps(obj, default=str, ensure_ascii=False)
                print(f"Fallback to standard json: {fallback_result}")
                return fallback_result.encode()
            except Exception as fallback_error:
                print(f"Fallback also failed: {fallback_error}")
                return b'{"error": "serialization_failed"}'
    
    def safe_loads(data: bytes | str) -> Any:
        """安全なorjson.loads"""
        try:
            return orjson.loads(data)
        except orjson.JSONDecodeError as e:
            print(f"Deserialization failed: {e}")
            # フォールバック: 標準jsonを使用
            try:
                if isinstance(data, bytes):
                    data = data.decode('utf-8', 'replace')
                return json.loads(data)
            except Exception as fallback_error:
                print(f"Fallback also failed: {fallback_error}")
                return {"error": "deserialization_failed"}
    
    def validate_json_compatibility(obj: Any) -> bool:
        """JSON互換性の検証"""
        try:
            # orjsonでテスト
            orjson_result = orjson.dumps(obj)
            orjson_parsed = orjson.loads(orjson_result)
            
            # 標準jsonでテスト（可能な場合）
            try:
                json_result = json.dumps(obj, default=str, ensure_ascii=False)
                json_parsed = json.loads(json_result)
                print("Both orjson and standard json succeeded")
                return True
            except Exception:
                print("orjson succeeded, but standard json failed")
                return True
                
        except Exception as e:
            print(f"JSON compatibility check failed: {e}")
            return False
    
    # テストデータでの検証
    test_cases = [
        {"simple": "data"},
        {1, 2, 3},  # set - 失敗予定
        datetime.datetime.now(),  # datetime - orjsonで成功
        {"valid": True, "number": 42}
    ]
    
    for i, test_case in enumerate(test_cases):
        print(f"\nTest case {i + 1}: {type(test_case).__name__}")
        
        # 安全な操作テスト
        result = safe_dumps(test_case)
        parsed = safe_loads(result)
        
        # 互換性チェック
        is_compatible = validate_json_compatibility(test_case)
        print(f"Compatibility: {is_compatible}")

if __name__ == "__main__":
    print("=== Comprehensive Error Handling ===")
    comprehensive_error_handling()
    
    print("\n=== Debug Helper Functions ===") 
    debug_helper_functions()