Cerberus

バリデーションライブラリPythonスキーマ辞書型軽量拡張可能

ライブラリ

Cerberus

概要

CerberusはPython専用の軽量で拡張可能なデータバリデーションライブラリです。「箱から出してすぐに使える」強力ながらシンプルなバリデーション機能を提供し、外部依存関係を持たない純粋なPythonライブラリとして設計されています。辞書ベースのスキーマ定義を採用し、YAML、JSON等の設定ファイルとの統合が容易です。2025年現在もセマンティックバージョニングに従って維持され、CPythonとPyPyの幅広いバージョンに対応した安定したバリデーションソリューションです。

詳細

Cerberus 1.3系は2025年現在の最新安定版で、セマンティックバージョニングに従った継続的メンテナンスが行われています。Pythonの標準型(dict、list、string等)を使用してスキーマを構築するため、PyYAMLやJSONなど様々な形式でスキーマを定義可能です。他のバリデーションツールと異なり、最初のエラーで停止せず、ドキュメント全体を処理してからFalseを返し、errors()メソッドで問題の一覧を取得できます。Pydanticと類似した機能を持ちながら、より軽量でシンプルな設計思想を採用しています。

主な特徴

  • 軽量設計: 外部依存関係なしの純粋Pythonライブラリ
  • 辞書ベーススキーマ: 直感的で読みやすいスキーマ定義
  • 拡張可能性: クラスベースと関数ベースのカスタムバリデーター対応
  • 包括的エラー処理: 全体処理後の詳細なエラーレポート
  • 設定ファイル統合: YAML、JSON等との自然な統合
  • マルチプラットフォーム: CPython、PyPyの幅広いバージョン対応

メリット・デメリット

メリット

  • 依存関係がなくシンプルで軽量
  • 辞書型スキーマによる直感的な定義方法
  • YAML/JSON設定ファイルとの高い親和性
  • カスタムバリデーターの実装が容易
  • 全エラー収集による包括的なバリデーション結果
  • セマンティックバージョニングによる安定性

デメリット

  • Pydanticより機能が限定的
  • 型ヒントサポートの不足
  • 真偽値返却方式で例外ベース処理に不向き
  • 大規模プロジェクトでのスキーマ管理複雑化
  • IDE支援やオートコンプリートの制限
  • モダンPythonの typing システムとの統合不足

参考ページ

書き方の例

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

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

# 依存関係なしの軽量ライブラリです
# Python 3.7以上が必要

基本的なスキーマ定義とバリデーション

from cerberus import Validator

# 基本的なスキーマ定義
schema = {
    'name': {
        'type': 'string',
        'required': True,
        'minlength': 2,
        'maxlength': 50
    },
    'age': {
        'type': 'integer',
        'required': True,
        'min': 0,
        'max': 150
    },
    'email': {
        'type': 'string',
        'required': True,
        'regex': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    },
    'active': {
        'type': 'boolean',
        'default': True
    },
    'score': {
        'type': 'float',
        'nullable': True,
        'min': 0.0,
        'max': 100.0
    }
}

# バリデーターのインスタンス作成
v = Validator(schema)

# 正常なデータのバリデーション
valid_document = {
    'name': '田中太郎',
    'age': 30,
    'email': '[email protected]',
    'active': True,
    'score': 85.5
}

if v.validate(valid_document):
    print("バリデーション成功")
    print("正規化されたデータ:", v.normalized(valid_document))
else:
    print("バリデーションエラー:", v.errors)

# 無効なデータのバリデーション
invalid_document = {
    'name': '',  # 最小長違反
    'age': -5,   # 最小値違反
    'email': 'invalid-email',  # 正規表現違反
    'score': 150.0  # 最大値違反
}

if v.validate(invalid_document):
    print("バリデーション成功")
else:
    print("バリデーションエラー:")
    for field, errors in v.errors.items():
        print(f"  {field}: {errors}")

# デフォルト値の適用
minimal_document = {
    'name': '山田花子',
    'age': 25,
    'email': '[email protected]'
}

if v.validate(minimal_document):
    normalized = v.normalized(minimal_document)
    print("デフォルト値適用後:", normalized)
    # 結果: active が True に設定される

ネストしたスキーマと複雑なバリデーション

# ネストしたオブジェクトのスキーマ
address_schema = {
    'street': {
        'type': 'string',
        'required': True,
        'minlength': 5
    },
    'city': {
        'type': 'string',
        'required': True
    },
    'postal_code': {
        'type': 'string',
        'required': True,
        'regex': r'^\d{3}-\d{4}$'
    },
    'country': {
        'type': 'string',
        'allowed': ['Japan', 'USA', 'UK', 'Canada']
    }
}

# メインスキーマでのネスト
user_schema = {
    'personal_info': {
        'type': 'dict',
        'required': True,
        'schema': {
            'first_name': {'type': 'string', 'required': True, 'minlength': 1},
            'last_name': {'type': 'string', 'required': True, 'minlength': 1},
            'birth_date': {
                'type': 'datetime',
                'required': True
            }
        }
    },
    'address': {
        'type': 'dict',
        'required': True,
        'schema': address_schema
    },
    'phones': {
        'type': 'list',
        'required': False,
        'schema': {
            'type': 'dict',
            'schema': {
                'type': {
                    'type': 'string',
                    'allowed': ['mobile', 'home', 'work']
                },
                'number': {
                    'type': 'string',
                    'regex': r'^\d{3}-\d{4}-\d{4}$'
                }
            }
        }
    },
    'tags': {
        'type': 'list',
        'required': False,
        'schema': {
            'type': 'string',
            'minlength': 1,
            'maxlength': 20
        },
        'minlength': 0,
        'maxlength': 10
    }
}

# 複雑なドキュメントのバリデーション
from datetime import datetime

complex_document = {
    'personal_info': {
        'first_name': '太郎',
        'last_name': '田中',
        'birth_date': datetime(1990, 5, 15)
    },
    'address': {
        'street': '渋谷区道玄坂1-2-3',
        'city': '東京',
        'postal_code': '150-0043',
        'country': 'Japan'
    },
    'phones': [
        {'type': 'mobile', 'number': '090-1234-5678'},
        {'type': 'home', 'number': '03-1234-5678'}
    ],
    'tags': ['developer', 'python', 'web']
}

validator = Validator(user_schema)

if validator.validate(complex_document):
    print("複雑なドキュメントのバリデーション成功")
else:
    print("バリデーションエラー:")
    for field, errors in validator.errors.items():
        print(f"  {field}: {errors}")

# リストの各要素バリデーション
scores_schema = {
    'student_scores': {
        'type': 'list',
        'required': True,
        'minlength': 1,
        'schema': {
            'type': 'dict',
            'schema': {
                'student_id': {'type': 'string', 'required': True, 'regex': r'^S\d{6}$'},
                'name': {'type': 'string', 'required': True},
                'scores': {
                    'type': 'dict',
                    'required': True,
                    'schema': {
                        'math': {'type': 'integer', 'min': 0, 'max': 100},
                        'english': {'type': 'integer', 'min': 0, 'max': 100},
                        'science': {'type': 'integer', 'min': 0, 'max': 100}
                    }
                }
            }
        }
    }
}

scores_document = {
    'student_scores': [
        {
            'student_id': 'S123456',
            'name': '田中太郎',
            'scores': {'math': 85, 'english': 90, 'science': 78}
        },
        {
            'student_id': 'S123457',
            'name': '山田花子',
            'scores': {'math': 92, 'english': 88, 'science': 95}
        }
    ]
}

scores_validator = Validator(scores_schema)
if scores_validator.validate(scores_document):
    print("成績データのバリデーション成功")

カスタムバリデーターと高度な機能

from cerberus import Validator
import re
from datetime import datetime, date

# カスタムバリデータークラス
class CustomValidator(Validator):
    def _validate_is_even(self, is_even, field, value):
        """偶数かどうかをチェックするカスタムルール
        
        The rule's arguments are validated against this schema:
        {'type': 'boolean'}
        """
        if is_even and value % 2 != 0:
            self._error(field, f"{value} は偶数である必要があります")
    
    def _validate_future_date(self, future_date, field, value):
        """未来の日付かどうかをチェック
        
        The rule's arguments are validated against this schema:
        {'type': 'boolean'}
        """
        if future_date and isinstance(value, (date, datetime)):
            if value <= date.today():
                self._error(field, "未来の日付である必要があります")
    
    def _validate_strong_password(self, strong_password, field, value):
        """強いパスワードかどうかをチェック
        
        The rule's arguments are validated against this schema:
        {'type': 'boolean'}
        """
        if strong_password and isinstance(value, str):
            if len(value) < 8:
                self._error(field, "パスワードは8文字以上である必要があります")
            if not re.search(r'[A-Z]', value):
                self._error(field, "パスワードには大文字が含まれている必要があります")
            if not re.search(r'[a-z]', value):
                self._error(field, "パスワードには小文字が含まれている必要があります")
            if not re.search(r'\d', value):
                self._error(field, "パスワードには数字が含まれている必要があります")
            if not re.search(r'[!@#$%^&*(),.?":{}|<>]', value):
                self._error(field, "パスワードには特殊文字が含まれている必要があります")

# カスタムルールを使用するスキーマ
custom_schema = {
    'user_id': {
        'type': 'integer',
        'required': True,
        'is_even': True  # カスタムルール
    },
    'password': {
        'type': 'string',
        'required': True,
        'strong_password': True  # カスタムルール
    },
    'appointment_date': {
        'type': 'date',
        'required': True,
        'future_date': True  # カスタムルール
    },
    'username': {
        'type': 'string',
        'required': True,
        'minlength': 3,
        'maxlength': 20,
        'regex': r'^[a-zA-Z0-9_]+$'
    }
}

# カスタムバリデーターの使用
custom_validator = CustomValidator(custom_schema)

# テストデータ
test_data = {
    'user_id': 12,  # 偶数
    'password': 'SecurePass123!',  # 強いパスワード
    'appointment_date': date(2025, 12, 31),  # 未来の日付
    'username': 'user_123'
}

if custom_validator.validate(test_data):
    print("カスタムバリデーション成功")
else:
    print("カスタムバリデーションエラー:")
    for field, errors in custom_validator.errors.items():
        print(f"  {field}: {errors}")

# 関数ベースのカスタムバリデーター
def validate_japanese_phone(field, value, error):
    """日本の電話番号形式をチェック"""
    if not re.match(r'^(070|080|090)-\d{4}-\d{4}$|^0\d{1,4}-\d{1,4}-\d{4}$', value):
        error(field, '日本の電話番号形式で入力してください(例:090-1234-5678)')

# 関数ベースバリデーターの登録
phone_schema = {
    'phone': {
        'type': 'string',
        'required': True,
        'check_with': validate_japanese_phone
    }
}

phone_validator = Validator(phone_schema)
phone_data = {'phone': '090-1234-5678'}

if phone_validator.validate(phone_data):
    print("電話番号バリデーション成功")

条件付きバリデーションとスキーマの動的生成

# 条件付きバリデーション(depend on other fields)
conditional_schema = {
    'account_type': {
        'type': 'string',
        'required': True,
        'allowed': ['personal', 'business', 'premium']
    },
    'company_name': {
        'type': 'string',
        'required': False,
        'dependencies': 'account_type'  # account_typeが存在する場合のみ必要
    },
    'tax_id': {
        'type': 'string',
        'required': False,
        'dependencies': ['account_type', 'company_name']
    },
    'credit_limit': {
        'type': 'integer',
        'required': False,
        'min': 1000,
        'dependencies': 'account_type'
    }
}

class ConditionalValidator(Validator):
    def _validate_dependencies(self, dependencies, field, value):
        """条件付き必須フィールドのチェック"""
        if isinstance(dependencies, str):
            dependencies = [dependencies]
        
        document = self.document
        for dep in dependencies:
            if dep in document:
                if dep == 'account_type':
                    if document['account_type'] == 'business':
                        if field == 'company_name' and not value:
                            self._error(field, 'ビジネスアカウントでは会社名が必要です')
                        elif field == 'tax_id' and not value:
                            self._error(field, 'ビジネスアカウントでは税務IDが必要です')
                    elif document['account_type'] == 'premium':
                        if field == 'credit_limit' and not value:
                            self._error(field, 'プレミアムアカウントでは与信限度額が必要です')

# スキーマの動的生成
def generate_form_schema(form_type):
    """フォームタイプに応じてスキーマを動的生成"""
    base_schema = {
        'name': {'type': 'string', 'required': True, 'minlength': 1},
        'email': {'type': 'string', 'required': True, 'regex': r'^[^@]+@[^@]+\.[^@]+$'}
    }
    
    if form_type == 'registration':
        base_schema.update({
            'password': {
                'type': 'string',
                'required': True,
                'minlength': 8
            },
            'confirm_password': {
                'type': 'string',
                'required': True
            },
            'age': {
                'type': 'integer',
                'required': True,
                'min': 13
            }
        })
    elif form_type == 'profile':
        base_schema.update({
            'bio': {
                'type': 'string',
                'required': False,
                'maxlength': 500
            },
            'website': {
                'type': 'string',
                'required': False,
                'regex': r'^https?://.+'
            }
        })
    elif form_type == 'contact':
        base_schema.update({
            'subject': {
                'type': 'string',
                'required': True,
                'minlength': 5
            },
            'message': {
                'type': 'string',
                'required': True,
                'minlength': 10
            }
        })
    
    return base_schema

# 動的スキーマの使用例
registration_schema = generate_form_schema('registration')
profile_schema = generate_form_schema('profile')
contact_schema = generate_form_schema('contact')

# バリデーション実行
registration_data = {
    'name': '田中太郎',
    'email': '[email protected]',
    'password': 'SecurePass123',
    'confirm_password': 'SecurePass123',
    'age': 25
}

reg_validator = Validator(registration_schema)
if reg_validator.validate(registration_data):
    print("登録フォームバリデーション成功")

エラーハンドリングとYAML/JSON統合

import yaml
import json
from cerberus import Validator

# YAML形式でのスキーマ定義
yaml_schema_str = """
user:
  type: dict
  required: true
  schema:
    name:
      type: string
      required: true
      minlength: 2
      maxlength: 50
    age:
      type: integer
      required: true
      min: 0
      max: 150
    email:
      type: string
      required: true
      regex: '^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$'
    preferences:
      type: dict
      required: false
      schema:
        newsletter:
          type: boolean
          default: false
        language:
          type: string
          allowed: ['ja', 'en', 'fr', 'de']
          default: 'ja'
"""

# YAMLからスキーマを読み込み
yaml_schema = yaml.safe_load(yaml_schema_str)

# JSON形式でのテストデータ
json_data_str = '''
{
  "user": {
    "name": "田中太郎",
    "age": 30,
    "email": "[email protected]",
    "preferences": {
      "newsletter": true,
      "language": "ja"
    }
  }
}
'''

test_data = json.loads(json_data_str)

# 包括的なエラーハンドリング
class DetailedValidator(Validator):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.validation_summary = {
            'total_fields': 0,
            'valid_fields': 0,
            'error_fields': 0,
            'errors_by_type': {}
        }
    
    def validate(self, document, schema=None, update=False, normalize=True):
        result = super().validate(document, schema, update, normalize)
        self._generate_summary()
        return result
    
    def _generate_summary(self):
        """バリデーション結果のサマリーを生成"""
        if hasattr(self, 'document'):
            self.validation_summary['total_fields'] = len(self._flatten_dict(self.document))
        
        if self.errors:
            error_dict = self._flatten_dict(self.errors)
            self.validation_summary['error_fields'] = len(error_dict)
            self.validation_summary['valid_fields'] = (
                self.validation_summary['total_fields'] - 
                self.validation_summary['error_fields']
            )
            
            # エラータイプ別集計
            for field_path, error_list in error_dict.items():
                for error in error_list:
                    if 'required' in error:
                        self.validation_summary['errors_by_type']['required'] = (
                            self.validation_summary['errors_by_type'].get('required', 0) + 1
                        )
                    elif 'type' in error:
                        self.validation_summary['errors_by_type']['type'] = (
                            self.validation_summary['errors_by_type'].get('type', 0) + 1
                        )
                    elif any(keyword in error for keyword in ['min', 'max', 'minlength', 'maxlength']):
                        self.validation_summary['errors_by_type']['range'] = (
                            self.validation_summary['errors_by_type'].get('range', 0) + 1
                        )
                    else:
                        self.validation_summary['errors_by_type']['other'] = (
                            self.validation_summary['errors_by_type'].get('other', 0) + 1
                        )
        else:
            self.validation_summary['valid_fields'] = self.validation_summary['total_fields']
    
    def _flatten_dict(self, d, parent_key='', sep='.'):
        """ネストした辞書をフラット化"""
        items = []
        for k, v in d.items():
            new_key = f"{parent_key}{sep}{k}" if parent_key else k
            if isinstance(v, dict):
                items.extend(self._flatten_dict(v, new_key, sep=sep).items())
            else:
                items.append((new_key, v))
        return dict(items)
    
    def get_detailed_report(self):
        """詳細なバリデーションレポートを生成"""
        report = {
            'validation_result': len(self.errors) == 0,
            'summary': self.validation_summary,
            'errors': self.errors if self.errors else None,
            'recommendations': []
        }
        
        # 推奨事項の生成
        if self.validation_summary['errors_by_type'].get('required', 0) > 0:
            report['recommendations'].append('必須フィールドの入力を確認してください')
        if self.validation_summary['errors_by_type'].get('type', 0) > 0:
            report['recommendations'].append('データ型が正しいか確認してください')
        if self.validation_summary['errors_by_type'].get('range', 0) > 0:
            report['recommendations'].append('値の範囲や長さ制限を確認してください')
        
        return report

# 詳細バリデーターの使用
detailed_validator = DetailedValidator(yaml_schema)

if detailed_validator.validate(test_data):
    print("YAML/JSONバリデーション成功")
    print("正規化されたデータ:", detailed_validator.normalized(test_data))
else:
    report = detailed_validator.get_detailed_report()
    print("詳細バリデーションレポート:")
    print(json.dumps(report, indent=2, ensure_ascii=False))

# 設定ファイルバリデーションの実用例
def validate_config_file(config_path, schema_path):
    """設定ファイルのバリデーション"""
    try:
        with open(schema_path, 'r', encoding='utf-8') as f:
            schema = yaml.safe_load(f)
        
        with open(config_path, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f)
        
        validator = DetailedValidator(schema)
        
        if validator.validate(config):
            return {
                'valid': True,
                'config': validator.normalized(config),
                'message': '設定ファイルは有効です'
            }
        else:
            return {
                'valid': False,
                'errors': validator.errors,
                'report': validator.get_detailed_report(),
                'message': '設定ファイルにエラーがあります'
            }
    
    except Exception as e:
        return {
            'valid': False,
            'error': str(e),
            'message': 'ファイルの読み込みまたは解析でエラーが発生しました'
        }