marshmallow

バリデーションライブラリPythonシリアライゼーションスキーマオブジェクト変換軽量

ライブラリ

marshmallow

概要

marshmallowは、複雑なオブジェクトとシンプルなPythonデータ型の相互変換を行う軽量ライブラリです。データのバリデーション、シリアライゼーション、デシリアライゼーションの3つの主要機能を提供し、Python Webアプリケーション開発の標準的なツールとして広く採用されています。2025年現在、バージョン4.0.0が最新で、Flask、FastAPI、SQLAlchemy等との統合により、エンタープライズレベルのPythonアプリケーションで重要な地位を占めています。Tidelift Subscriptionによる商用サポートも提供されています。

詳細

marshmallow 4.0系は2025年現在の最新メジャーバージョンで、スキーマベースのデータ処理アプローチを採用。入力データの検証、アプリケーションレベルオブジェクトへのデシリアライゼーション、プリミティブPython型へのシリアライゼーションという3段階処理により、HTTP APIでのJSON処理など、標準フォーマットでのデータレンダリングを効率化します。Pythonオブジェクト指向パラダイムでの強力な抽象化ソリューションを提供し、他言語のバリデーションライブラリと類似したスキーマ定義手法により、多くの開発者に支持されています。

主な特徴

3段階データ処理: バリデーション、デシリアライゼーション、シリアライゼーション
スキーマベース設計: 宣言的で再利用可能なスキーマ定義
フレームワーク統合: Flask、FastAPI、SQLAlchemy等との標準的統合
軽量アーキテクチャ: 最小限の依存関係で高いパフォーマンス
豊富なフィールド型: 内蔵フィールドと拡張可能なカスタムフィールド
商用サポート: Tideliftによるエンタープライズサポート

メリット・デメリット

メリット

Pythonエコシステムでの標準的地位と成熟度
Flask、FastAPI等主要フレームワークとの優れた統合
スキーマ定義による直感的で保守性の高いAPI
軽量でありながら豊富な機能セット
ネストしたオブジェクト構造の自然な処理
部分的シリアライゼーション（only/exclude）の柔軟性

デメリット

Pydanticと比較して型ヒントサポートが限定的
スキーマクラス定義の冗長性（ボイラープレート）
ランタイムでのパフォーマンスオーバーヘッド
複雑なバリデーションロジックでの設定の複雑化
モダンPythonの型システムとの統合に課題
エラーハンドリングの学習コストがやや高い

参考ページ

書き方の例

インストールと基本セットアップ

# marshmallowのインストール
pip install marshmallow
poetry add marshmallow
pipenv install marshmallow

# エクストラ機能（オプション）
pip install marshmallow[dev]  # 開発用依存関係
pip install marshmallow[reco] # 推奨依存関係

# Python 3.8以上が必要
# 軽量で依存関係最小限

基本的なスキーマ定義とシリアライゼーション

from marshmallow import Schema, fields
from dataclasses import dataclass
from datetime import datetime
from pprint import pprint

# データクラスの定義
@dataclass
class User:
    name: str
    email: str
    created_at: datetime

# marshmallowスキーマの定義
class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

# ユーザーオブジェクトの作成
user = User(
    name="田中太郎",
    email="[email protected]",
    created_at=datetime.now()
)

# シリアライゼーション（オブジェクト → 辞書）
schema = UserSchema()
result = schema.dump(user)
pprint(result)
# {'name': '田中太郎',
#  'email': '[email protected]', 
#  'created_at': '2025-06-22T10:30:45.123456'}

# 複数オブジェクトのシリアライゼーション
users = [
    User("山田花子", "[email protected]", datetime.now()),
    User("佐藤次郎", "[email protected]", datetime.now())
]
schema_many = UserSchema(many=True)
results = schema_many.dump(users)
print(f"ユーザー数: {len(results)}")

# 部分的シリアライゼーション（フィールド選択）
summary_schema = UserSchema(only=("name", "email"))
summary = summary_schema.dump(user)
pprint(summary)
# {'name': '田中太郎', 'email': '[email protected]'}

# フィールド除外
no_timestamp_schema = UserSchema(exclude=("created_at",))
no_timestamp = no_timestamp_schema.dump(user)
pprint(no_timestamp)
# {'name': '田中太郎', 'email': '[email protected]'}

デシリアライゼーションとバリデーション

from marshmallow import Schema, fields, ValidationError
from pprint import pprint

class UserSchema(Schema):
    name = fields.Str(required=True)
    email = fields.Email(required=True)
    age = fields.Int(validate=lambda x: 0 <= x <= 150)
    created_at = fields.DateTime()

# 有効なデータのデシリアライゼーション
user_data = {
    "name": "田中太郎",
    "email": "[email protected]",
    "age": 30,
    "created_at": "2025-06-22T10:30:45.123456"
}

schema = UserSchema()
try:
    result = schema.load(user_data)
    pprint(result)
    # {'name': '田中太郎',
    #  'email': '[email protected]',
    #  'age': 30,
    #  'created_at': datetime.datetime(2025, 6, 22, 10, 30, 45, 123456)}
    print("デシリアライゼーション成功")
except ValidationError as err:
    print("バリデーションエラー:", err.messages)

# 無効なデータでのバリデーションエラー
invalid_data = {
    "name": "",  # 空文字
    "email": "invalid-email",  # 無効なメール形式
    "age": 200,  # 範囲外の値
    "created_at": "invalid-date"  # 無効な日付形式
}

try:
    result = schema.load(invalid_data)
except ValidationError as err:
    print("バリデーションエラー:")
    for field, errors in err.messages.items():
        print(f"  {field}: {errors}")
    
    # 有効だったデータも取得可能
    if err.valid_data:
        print("有効だったデータ:", err.valid_data)

# バリデーションのみ（エラーチェック）
validation_result = schema.validate(invalid_data)
if validation_result:
    print("バリデーションエラー発見:", validation_result)
else:
    print("バリデーション成功")

# 部分的デシリアライゼーション
partial_data = {"name": "田中太郎"}
try:
    result = schema.load(partial_data, partial=True)
    print("部分的デシリアライゼーション成功:", result)
except ValidationError as err:
    print("エラー:", err.messages)

ネストしたスキーマと複雑な構造

from marshmallow import Schema, fields
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

# 複雑なデータ構造の定義
@dataclass
class Address:
    street: str
    city: str
    postal_code: str
    country: str

@dataclass
class Company:
    name: str
    industry: str

@dataclass
class User:
    name: str
    email: str
    address: Address
    company: Optional[Company]
    skills: List[str]
    created_at: datetime

# ネストしたスキーマの定義
class AddressSchema(Schema):
    street = fields.Str(required=True)
    city = fields.Str(required=True)
    postal_code = fields.Str(required=True)
    country = fields.Str(required=True)

class CompanySchema(Schema):
    name = fields.Str(required=True)
    industry = fields.Str(required=True)

class UserSchema(Schema):
    name = fields.Str(required=True)
    email = fields.Email(required=True)
    address = fields.Nested(AddressSchema, required=True)
    company = fields.Nested(CompanySchema, allow_none=True)
    skills = fields.List(fields.Str(), required=True)
    created_at = fields.DateTime()

# 複雑なデータの作成
complex_data = {
    "name": "田中太郎",
    "email": "[email protected]",
    "address": {
        "street": "渋谷区道玄坂1-2-3",
        "city": "東京",
        "postal_code": "150-0043",
        "country": "Japan"
    },
    "company": {
        "name": "株式会社サンプル",
        "industry": "IT"
    },
    "skills": ["Python", "JavaScript", "Go"],
    "created_at": "2025-06-22T10:30:45.123456"
}

schema = UserSchema()

# 複雑なデータのデシリアライゼーション
try:
    result = schema.load(complex_data)
    print("複雑なデータのデシリアライゼーション成功")
    print(f"住所: {result['address']['city']}")
    print(f"スキル数: {len(result['skills'])}")
except ValidationError as err:
    print("エラー:", err.messages)

# ネストしたフィールドの部分的シリアライゼーション
address_only_schema = UserSchema(only=("name", "address.city", "address.country"))
user_obj = User(
    name="田中太郎",
    email="[email protected]",
    address=Address("渋谷区道玄坂1-2-3", "東京", "150-0043", "Japan"),
    company=Company("株式会社サンプル", "IT"),
    skills=["Python", "JavaScript"],
    created_at=datetime.now()
)

# 特定のネストフィールドのみ取得
address_result = UserSchema(only=("name", "address")).dump(user_obj)
print("住所のみ:", address_result)

# リストでのネストオブジェクト
teams_data = {
    "name": "開発チーム",
    "members": [
        {"name": "田中太郎", "email": "[email protected]"},
        {"name": "山田花子", "email": "[email protected]"}
    ]
}

class TeamSchema(Schema):
    name = fields.Str(required=True)
    members = fields.List(fields.Nested(UserSchema(only=("name", "email"))))

team_schema = TeamSchema()
team_result = team_schema.load(teams_data)
print(f"チームメンバー数: {len(team_result['members'])}")

カスタムバリデーション関数と高度な機能

from marshmallow import Schema, fields, validates, validates_schema, ValidationError, post_load
import re
from datetime import datetime, date

class AdvancedUserSchema(Schema):
    username = fields.Str(required=True)
    password = fields.Str(required=True, load_only=True)  # シリアライゼーション時は除外
    confirm_password = fields.Str(required=True, load_only=True)
    email = fields.Email(required=True)
    phone = fields.Str(required=True)
    birth_date = fields.Date(required=True)
    created_at = fields.DateTime(dump_only=True, default=datetime.now)  # デシリアライゼーション時は無視

    @validates('username')
    def validate_username(self, value):
        """ユーザー名のカスタムバリデーション"""
        if len(value) < 3:
            raise ValidationError('ユーザー名は3文字以上である必要があります')
        if not re.match(r'^[a-zA-Z0-9_]+$', value):
            raise ValidationError('ユーザー名は英数字とアンダースコアのみ使用可能です')
        
        # 予約語チェック
        reserved_words = ['admin', 'root', 'system', 'test']
        if value.lower() in reserved_words:
            raise ValidationError('このユーザー名は予約されています')

    @validates('password')
    def validate_password(self, value):
        """パスワード強度チェック"""
        if len(value) < 8:
            raise ValidationError('パスワードは8文字以上である必要があります')
        
        # パスワード強度チェック
        if not re.search(r'[A-Z]', value):
            raise ValidationError('パスワードには大文字を含める必要があります')
        if not re.search(r'[a-z]', value):
            raise ValidationError('パスワードには小文字を含める必要があります')
        if not re.search(r'\d', value):
            raise ValidationError('パスワードには数字を含める必要があります')
        if not re.search(r'[!@#$%^&*(),.?":{}|<>]', value):
            raise ValidationError('パスワードには特殊文字を含める必要があります')

    @validates('phone')
    def validate_phone(self, value):
        """日本の電話番号形式チェック"""
        # 日本の電話番号パターン
        patterns = [
            r'^0\d{1,4}-\d{1,4}-\d{4}$',  # 固定電話
            r'^(070|080|090)-\d{4}-\d{4}$',  # 携帯電話
        ]
        
        if not any(re.match(pattern, value) for pattern in patterns):
            raise ValidationError('有効な日本の電話番号形式で入力してください')

    @validates('birth_date')
    def validate_birth_date(self, value):
        """生年月日の妥当性チェック"""
        today = date.today()
        age = today.year - value.year - ((today.month, today.day) < (value.month, value.day))
        
        if age < 13:
            raise ValidationError('13歳以上である必要があります')
        if age > 120:
            raise ValidationError('有効な生年月日を入力してください')

    @validates_schema
    def validate_passwords_match(self, data, **kwargs):
        """スキーマレベルのバリデーション：パスワード一致確認"""
        if 'password' in data and 'confirm_password' in data:
            if data['password'] != data['confirm_password']:
                raise ValidationError({'confirm_password': ['パスワードが一致しません']})

    @post_load
    def make_user(self, data, **kwargs):
        """デシリアライゼーション後の処理"""
        # confirm_passwordは内部処理用なので削除
        data.pop('confirm_password', None)
        # 作成日時を追加
        data['created_at'] = datetime.now()
        return data

# カスタムバリデーション関数
def validate_email_domain(email):
    """特定ドメインのメールアドレスのみ許可"""
    allowed_domains = ['example.com', 'company.com', 'gmail.com']
    domain = email.split('@')[1]
    if domain not in allowed_domains:
        raise ValidationError(f'許可されていないドメインです。許可ドメイン: {", ".join(allowed_domains)}')

class RestrictedUserSchema(AdvancedUserSchema):
    email = fields.Email(required=True, validate=validate_email_domain)

# 高度なバリデーションのテスト
advanced_data = {
    "username": "tanaka_user",
    "password": "SecurePass123!",
    "confirm_password": "SecurePass123!",
    "email": "[email protected]",
    "phone": "090-1234-5678",
    "birth_date": "1990-05-15"
}

schema = AdvancedUserSchema()

try:
    result = schema.load(advanced_data)
    print("高度なバリデーション成功:")
    print(f"ユーザー名: {result['username']}")
    print(f"作成日時: {result['created_at']}")
    # パスワードは含まれない（load_only）
    print("パスワード含まれていない:", 'password' not in result)
except ValidationError as err:
    print("バリデーションエラー:", err.messages)

# 無効なデータでのテスト
invalid_advanced_data = {
    "username": "ad",  # 短すぎる
    "password": "weak",  # 弱いパスワード
    "confirm_password": "different",  # 異なるパスワード
    "email": "[email protected]",  # 許可されていないドメイン
    "phone": "123-456-789",  # 無効な電話番号形式
    "birth_date": "2020-01-01"  # 年齢制限に引っかかる
}

try:
    result = RestrictedUserSchema().load(invalid_advanced_data)
except ValidationError as err:
    print("\n複数エラーのキャッチ:")
    for field, errors in err.messages.items():
        print(f"  {field}: {errors}")

WebフレームワークとAPI統合

from marshmallow import Schema, fields, ValidationError
from flask import Flask, request, jsonify
import json

# Flask統合例
app = Flask(__name__)

class CreateUserSchema(Schema):
    name = fields.Str(required=True)
    email = fields.Email(required=True)
    age = fields.Int(required=True, validate=lambda x: 18 <= x <= 100)

class UpdateUserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    age = fields.Int(validate=lambda x: 18 <= x <= 100)

class UserResponseSchema(Schema):
    id = fields.Int()
    name = fields.Str()
    email = fields.Email()
    age = fields.Int()
    created_at = fields.DateTime()

# バリデーション用デコレーター
def validate_json(schema_class):
    def decorator(f):
        def wrapper(*args, **kwargs):
            try:
                schema = schema_class()
                validated_data = schema.load(request.json)
                return f(validated_data, *args, **kwargs)
            except ValidationError as err:
                return jsonify({
                    'error': 'バリデーションエラー',
                    'messages': err.messages
                }), 400
        wrapper.__name__ = f.__name__
        return wrapper
    return decorator

@app.route('/users', methods=['POST'])
@validate_json(CreateUserSchema)
def create_user(validated_data):
    """ユーザー作成API"""
    # データベース保存処理（省略）
    user_data = {
        'id': 1,
        'created_at': '2025-06-22T10:30:45.123456',
        **validated_data
    }
    
    # レスポンスのシリアライゼーション
    response_schema = UserResponseSchema()
    return jsonify(response_schema.dump(user_data)), 201

@app.route('/users/<int:user_id>', methods=['PUT'])
@validate_json(UpdateUserSchema)
def update_user(validated_data, user_id):
    """ユーザー更新API"""
    # データベース更新処理（省略）
    user_data = {
        'id': user_id,
        'name': validated_data.get('name', '既存の名前'),
        'email': validated_data.get('email', '[email protected]'),
        'age': validated_data.get('age', 25),
        'created_at': '2025-06-22T10:30:45.123456'
    }
    
    response_schema = UserResponseSchema()
    return jsonify(response_schema.dump(user_data))

# FastAPI統合例（概念実装）
from typing import Optional

class FastAPIIntegration:
    """FastAPIでのmarshmallow統合パターン"""
    
    @staticmethod
    def validate_request_body(schema_class, data: dict):
        """リクエストボディのバリデーション"""
        try:
            schema = schema_class()
            return schema.load(data), None
        except ValidationError as err:
            return None, err.messages
    
    @staticmethod
    def serialize_response(schema_class, data):
        """レスポンスのシリアライゼーション"""
        schema = schema_class()
        return schema.dump(data)

# 使用例
def fastapi_create_user_handler(request_data: dict):
    """FastAPI風のハンドラー例"""
    validated_data, errors = FastAPIIntegration.validate_request_body(
        CreateUserSchema, request_data
    )
    
    if errors:
        return {"error": "Validation failed", "details": errors}, 400
    
    # ビジネスロジック処理
    user_data = {
        'id': 1,
        'created_at': '2025-06-22T10:30:45.123456',
        **validated_data
    }
    
    response_data = FastAPIIntegration.serialize_response(
        UserResponseSchema, user_data
    )
    
    return response_data, 200

# SQLAlchemy統合例
class SQLAlchemyIntegration:
    """SQLAlchemyモデルとの統合パターン"""
    
    def __init__(self, model_class, schema_class):
        self.model_class = model_class
        self.schema_class = schema_class
    
    def create_from_dict(self, data: dict):
        """辞書からモデルインスタンス作成"""
        schema = self.schema_class()
        validated_data = schema.load(data)
        return self.model_class(**validated_data)
    
    def serialize_model(self, model_instance):
        """モデルインスタンスをシリアライゼーション"""
        schema = self.schema_class()
        return schema.dump(model_instance)
    
    def bulk_serialize(self, model_instances):
        """複数インスタンスの一括シリアライゼーション"""
        schema = self.schema_class(many=True)
        return schema.dump(model_instances)

# ユースケース例
def api_workflow_example():
    """API処理ワークフローの例"""
    # 1. リクエストデータ
    request_data = {
        "name": "田中太郎",
        "email": "[email protected]",
        "age": 30
    }
    
    # 2. バリデーションとデシリアライゼーション
    create_schema = CreateUserSchema()
    try:
        validated_data = create_schema.load(request_data)
        print("バリデーション成功:", validated_data)
    except ValidationError as err:
        print("バリデーションエラー:", err.messages)
        return
    
    # 3. ビジネスロジック処理（ユーザー作成）
    user_data = {
        'id': 123,
        'created_at': '2025-06-22T10:30:45.123456',
        **validated_data
    }
    
    # 4. レスポンスのシリアライゼーション
    response_schema = UserResponseSchema()
    response_data = response_schema.dump(user_data)
    print("レスポンスデータ:", response_data)

if __name__ == "__main__":
    api_workflow_example()

高度なスキーマ設計とベストプラクティス

from marshmallow import Schema, fields, validates_schema, ValidationError, pre_load, post_load, pre_dump, post_dump
from marshmallow.validate import Length, Range, OneOf
from typing import Dict, Any
import json

class BaseTimestampSchema(Schema):
    """タイムスタンプを含む基底スキーマ"""
    created_at = fields.DateTime(dump_only=True)
    updated_at = fields.DateTime(dump_only=True)

class PaginationSchema(Schema):
    """ページネーション用スキーマ"""
    page = fields.Int(required=True, validate=Range(min=1))
    per_page = fields.Int(required=True, validate=Range(min=1, max=100))
    total = fields.Int(dump_only=True)
    pages = fields.Int(dump_only=True)

class SearchQuerySchema(Schema):
    """検索クエリ用スキーマ"""
    q = fields.Str(required=True, validate=Length(min=1, max=100))
    category = fields.Str(validate=OneOf(['tech', 'business', 'hobby']))
    sort_by = fields.Str(validate=OneOf(['created_at', 'updated_at', 'name']))
    sort_order = fields.Str(validate=OneOf(['asc', 'desc']), default='desc')

class DynamicFieldsSchema(Schema):
    """動的フィールド選択をサポートするスキーマ"""
    
    def __init__(self, only_fields=None, exclude_fields=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if only_fields:
            self.only = only_fields
        if exclude_fields:
            self.exclude = exclude_fields

class UserProfileSchema(BaseTimestampSchema, DynamicFieldsSchema):
    """ユーザープロファイルスキーマ"""
    id = fields.Int(dump_only=True)
    username = fields.Str(required=True)
    email = fields.Email(required=True)
    full_name = fields.Str(required=True)
    bio = fields.Str(allow_none=True)
    avatar_url = fields.Url(allow_none=True)
    is_active = fields.Bool(default=True)
    
    # プライベート情報（条件付きで除外）
    phone = fields.Str(allow_none=True)
    birth_date = fields.Date(allow_none=True)

    @pre_load
    def preprocess_input(self, data, **kwargs):
        """入力データの前処理"""
        # 文字列の正規化
        if 'full_name' in data:
            data['full_name'] = data['full_name'].strip()
        if 'username' in data:
            data['username'] = data['username'].lower().strip()
        return data

    @post_load
    def validate_and_transform(self, data, **kwargs):
        """ロード後の検証と変換"""
        # ユーザー名の重複チェック（実際のアプリではDB確認）
        existing_usernames = ['admin', 'root', 'test']
        if data.get('username') in existing_usernames:
            raise ValidationError({'username': ['このユーザー名は使用できません']})
        
        return data

    @pre_dump
    def prepare_for_serialization(self, data, **kwargs):
        """シリアライゼーション前の準備"""
        # コンテキストに基づいてプライベート情報を除外
        context = self.context or {}
        is_owner = context.get('is_owner', False)
        is_admin = context.get('is_admin', False)
        
        if not (is_owner or is_admin):
            # 他人のプロファイルの場合、プライベート情報を除外
            sensitive_fields = ['phone', 'birth_date']
            for field in sensitive_fields:
                if hasattr(data, field):
                    setattr(data, field, None)
                elif isinstance(data, dict) and field in data:
                    data.pop(field, None)
        
        return data

    @post_dump
    def add_computed_fields(self, data, **kwargs):
        """計算フィールドの追加"""
        # 年齢の計算（生年月日から）
        if data.get('birth_date'):
            from datetime import date
            birth_date = data['birth_date']
            if isinstance(birth_date, str):
                birth_date = date.fromisoformat(birth_date)
            today = date.today()
            age = today.year - birth_date.year - ((today.month, today.day) < (birth_date.month, birth_date.day))
            data['age'] = age
        
        return data

class BlogPostSchema(BaseTimestampSchema):
    """ブログ投稿スキーマ"""
    id = fields.Int(dump_only=True)
    title = fields.Str(required=True, validate=Length(min=1, max=200))
    content = fields.Str(required=True, validate=Length(min=10))
    slug = fields.Str(dump_only=True)
    status = fields.Str(validate=OneOf(['draft', 'published', 'archived']), default='draft')
    author = fields.Nested(UserProfileSchema(only_fields=['id', 'username', 'full_name']))
    tags = fields.List(fields.Str(), default=list)
    view_count = fields.Int(dump_only=True, default=0)

    @validates_schema
    def validate_content_and_title(self, data, **kwargs):
        """タイトルとコンテンツの関連バリデーション"""
        title = data.get('title', '')
        content = data.get('content', '')
        
        # タイトルが短すぎる場合のコンテンツ長制限
        if len(title) < 10 and len(content) < 100:
            raise ValidationError('短いタイトルの場合、コンテンツは100文字以上である必要があります')

class ApiResponseSchema(Schema):
    """API レスポンス用の統一スキーマ"""
    success = fields.Bool(default=True)
    message = fields.Str()
    data = fields.Raw()
    pagination = fields.Nested(PaginationSchema, allow_none=True)
    
    @post_dump
    def remove_none_values(self, data, **kwargs):
        """None値を削除"""
        return {k: v for k, v in data.items() if v is not None}

class SchemaManager:
    """スキーマ管理とユーティリティクラス"""
    
    @staticmethod
    def create_api_response(data=None, message="成功", success=True, pagination=None):
        """API レスポンスの作成"""
        response_data = {
            'success': success,
            'message': message,
            'data': data,
            'pagination': pagination
        }
        return ApiResponseSchema().dump(response_data)
    
    @staticmethod
    def validate_and_serialize_list(schema_class, data_list, context=None):
        """リストデータのバリデーションとシリアライゼーション"""
        try:
            schema = schema_class(many=True, context=context)
            validated_data = schema.load(data_list)
            serialized_data = schema.dump(validated_data)
            return serialized_data, None
        except ValidationError as err:
            return None, err.messages
    
    @staticmethod
    def dynamic_schema_serialization(schema_class, data, only_fields=None, exclude_fields=None, context=None):
        """動的フィールド選択でのシリアライゼーション"""
        schema = schema_class(
            only_fields=only_fields,
            exclude_fields=exclude_fields,
            context=context
        )
        return schema.dump(data)

# 実践的な使用例
def advanced_schema_example():
    """高度なスキーマ使用例"""
    
    # ユーザーデータ
    user_data = {
        'username': '  TanakaUser  ',  # 前処理でトリムされる
        'email': '[email protected]',
        'full_name': '田中太郎',
        'bio': 'Python開発者です',
        'phone': '090-1234-5678',
        'birth_date': '1990-05-15'
    }
    
    # 1. 所有者としてのシリアライゼーション（プライベート情報含む）
    user_schema_owner = UserProfileSchema(context={'is_owner': True})
    try:
        validated_user = user_schema_owner.load(user_data)
        serialized_owner = user_schema_owner.dump(validated_user)
        print("所有者表示:", json.dumps(serialized_owner, indent=2, ensure_ascii=False))
    except ValidationError as err:
        print("バリデーションエラー:", err.messages)
    
    # 2. 他人としてのシリアライゼーション（プライベート情報除外）
    user_schema_public = UserProfileSchema(context={'is_owner': False})
    serialized_public = user_schema_public.dump(validated_user)
    print("\n公開表示:", json.dumps(serialized_public, indent=2, ensure_ascii=False))
    
    # 3. API レスポンスの作成
    response = SchemaManager.create_api_response(
        data=serialized_public,
        message="ユーザー情報を取得しました"
    )
    print("\nAPI レスポンス:", json.dumps(response, indent=2, ensure_ascii=False))
    
    # 4. 動的フィールド選択
    summary_data = SchemaManager.dynamic_schema_serialization(
        UserProfileSchema,
        validated_user,
        only_fields=['username', 'full_name', 'bio'],
        context={'is_owner': False}
    )
    print("\n要約表示:", json.dumps(summary_data, indent=2, ensure_ascii=False))

if __name__ == "__main__":
    advanced_schema_example()