Pythonのデータ検証ライブラリ「Cerberus」を使いこなそう

はじめに

皆さん、こんにちは！今日は、今回はデータ検証の強力な味方となる「Cerberus」というライブラリについてご紹介します。Cerberusは、その名前が示すように、ギリシャ神話の冥界の番犬にちなんで名付けられました。データの正確性を守る忠実な番犬のように、このライブラリはあなたのプログラムのデータ検証を確実に行ってくれます。

Cerberusは軽量で拡張性が高く、カスタム検証も簡単に行えるため、多くのPythonプロジェクトで重宝されています。それでは、Cerberusの基本から応用まで、15の章に分けて詳しく見ていきましょう。

第1章：Cerberusのインストールと基本設定

Cerberusを使い始めるには、まずインストールが必要です。幸い、pipを使えば簡単にインストールできます。ターミナルを開いて、以下のコマンドを実行しましょう。

pip install cerberus

インストールが完了したら、Pythonスクリプトで以下のようにインポートします。

from cerberus import Validator

# Validatorインスタンスの作成
v = Validator()

# これで、Cerberusを使う準備が整いました！

Cerberusの魅力は、このシンプルな準備だけで強力なデータ検証が始められることです。次の章から、実際の使い方を見ていきましょう。

第2章：基本的なスキーマ定義

Cerberusの心臓部とも言えるのが「スキーマ」です。スキーマとは、検証したいデータの構造とルールを定義したものです。まずは、シンプルなスキーマを作ってみましょう。

from cerberus import Validator

# スキーマの定義
schema = {
    'name': {'type': 'string'},
    'age': {'type': 'integer', 'min': 0, 'max': 150},
    'email': {'type': 'string', 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'}
}

# Validatorインスタンスの作成
v = Validator(schema)

# データの検証
data = {'name': '山田太郎', 'age': 30, 'email': '[e-mail]'}
if v.validate(data):
    print("データは有効です")
else:
    print("エラー:", v.errors)

このスキーマでは、名前は文字列、年齢は0から150までの整数、メールアドレスは正規表現にマッチする文字列と定義しています。Cerberusの強みは、このように直感的にデータの制約を表現できることです。

第3章：複雑なデータ構造の検証

実際のアプリケーションでは、もっと複雑なデータ構造を扱うことが多いでしょう。Cerberusは、ネストされた辞書やリストも簡単に検証できます。

from cerberus import Validator

schema = {
    'user': {
        'type': 'dict',
        'schema': {
            'name': {'type': 'string'},
            'age': {'type': 'integer', 'min': 0},
            'hobbies': {
                'type': 'list',
                'schema': {'type': 'string'}
            }
        }
    },
    'products': {
        'type': 'list',
        'schema': {
            'type': 'dict',
            'schema': {
                'name': {'type': 'string'},
                'price': {'type': 'float', 'min': 0}
            }
        }
    }
}

v = Validator(schema)

data = {
    'user': {
        'name': '佐藤花子',
        'age': 25,
        'hobbies': ['読書', '旅行', 'プログラミング']
    },
    'products': [
        {'name': '本', 'price': 1500.0},
        {'name': 'ノートPC', 'price': 150000.0}
    ]
}

if v.validate(data):
    print("複雑なデータ構造も問題なく検証できました")
else:
    print("エラー:", v.errors)

このように、Cerberusは複雑なデータ構造も直感的に定義し、検証することができます。

第4章：カスタムバリデーションルールの作成

Cerberusの標準ルールだけでは足りない場合、カスタムルールを作成することができます。例えば、日本の郵便番号を検証する関数を作ってみましょう。

from cerberus import Validator

def validate_japanese_postcode(field, value, error):
    if not (len(value) == 7 and value.isdigit()):
        error(field, "有効な郵便番号ではありません")

schema = {
    'name': {'type': 'string'},
    'postcode': {
        'type': 'string',
        'check_with': validate_japanese_postcode
    }
}

class CustomValidator(Validator):
    def _check_with_validate_japanese_postcode(self, field, value):
        validate_japanese_postcode(field, value, self.error)

v = CustomValidator(schema)

data1 = {'name': '鈴木一郎', 'postcode': '1234567'}
data2 = {'name': '田中二郎', 'postcode': '123-4567'}

print(v.validate(data1))  # True
print(v.validate(data2))  # False
print(v.errors)  # {'postcode': ['有効な郵便番号ではありません']}

このように、check_withキーを使ってカスタム関数を指定することで、独自の検証ルールを適用できます。

第5章：条件付きバリデーション

時には、あるフィールドの値に応じて他のフィールドの検証ルールを変更したいことがあります。Cerberusでは、dependenciesキーを使って条件付きバリデーションを実現できます。

from cerberus import Validator

schema = {
    'payment_method': {'type': 'string', 'allowed': ['現金', 'クレジットカード', '銀行振込']},
    'credit_card_number': {
        'type': 'string',
        'dependencies': {'payment_method': 'クレジットカード'},
        'regex': '^[0-9]{16}$'
    },
    'bank_account': {
        'type': 'string',
        'dependencies': {'payment_method': '銀行振込'},
        'regex': '^[0-9]{7}$'
    }
}

v = Validator(schema)

data1 = {'payment_method': 'クレジットカード', 'credit_card_number': '1234567890123456'}
data2 = {'payment_method': '銀行振込', 'bank_account': '1234567'}
data3 = {'payment_method': '現金'}
data4 = {'payment_method': 'クレジットカード', 'bank_account': '1234567'}

print(v.validate(data1))  # True
print(v.validate(data2))  # True
print(v.validate(data3))  # True
print(v.validate(data4))  # False
print(v.errors)  # {'credit_card_number': ['required field']}

この例では、支払い方法に応じて必要なフィールドが変わります。Cerberusはこのような複雑な条件も簡単に表現できます。

第6章：デフォルト値の設定

データの検証時に、特定のフィールドが存在しない場合にデフォルト値を設定したいことがあります。Cerberusでは、defaultキーを使ってこれを実現できます。

from cerberus import Validator

schema = {
    'username': {'type': 'string'},
    'email': {'type': 'string', 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'},
    'age': {'type': 'integer', 'min': 0, 'default': 20},
    'newsletter': {'type': 'boolean', 'default': False}
}

v = Validator(schema)

data = {'username': 'user123', 'email': '[e-mail]'}

# デフォルト値を適用して検証
if v.validate(data, update=True):
    print("検証成功。デフォルト値が適用されました。")
    print(v.document)
else:
    print("エラー:", v.errors)

この例では、ageとnewsletterフィールドにデフォルト値を設定しています。update=Trueオプションを使用することで、検証後にデフォルト値が適用されたデータをv.documentから取得できます。

第7章：コワーシブ（型変換）の使用

時には、入力データの型を自動的に変換したい場合があります。Cerberusの「コワーシブ」機能を使えば、これを簡単に実現できます。

from cerberus import Validator

def to_bool(value):
    if isinstance(value, bool):
        return value
    if value.lower() in ('yes', 'true', '1'):
        return True
    elif value.lower() in ('no', 'false', '0'):
        return False
    raise ValueError('Boolean値に変換できません')

schema = {
    'name': {'type': 'string'},
    'age': {'type': 'integer', 'coerce': int},
    'height': {'type': 'float', 'coerce': float},
    'is_student': {'type': 'boolean', 'coerce': to_bool}
}

v = Validator(schema)

data = {
    'name': '山田太郎',
    'age': '25',
    'height': '175.5',
    'is_student': 'yes'
}

if v.validate(data):
    print("データは有効で、型変換されました")
    print(v.document)
else:
    print("エラー:", v.errors)

この例では、ageを整数に、heightを浮動小数点数に、is_studentをブール値に自動変換しています。カスタム関数to_boolを使用して、文字列をブール値に変換する方法も示しています。

第8章：正規化（Normalization）の活用

データの検証だけでなく、正規化も行いたい場合があります。Cerberusの正規化機能を使えば、データを一定の形式に整えることができます。

from cerberus import Validator

def normalize_phone(phone):
    # ハイフンを削除し、先頭に'+'を付ける
    return '+' + ''.join(filter(str.isdigit, phone))

schema = {
    'name': {'type': 'string', 'coerce': str.strip},  # 前後の空白を削除
    'email': {'type': 'string', 'coerce': str.lower},  # 小文字に変換
    'phone': {'type': 'string', 'coerce': normalize_phone}  # 電話番号を正規化
}

v = Validator(schema)

data = {
    'name': '  佐藤花子  ',
    'email': '[e-mail]',
    'phone': '080-1234-5678'
}

if v.validate(data, normalize=True):
    print("データは正規化されました")
    print(v.document)
else:
    print("エラー:", v.errors)

この例では、名前の前後の空白を削除し、メールアドレスを小文字に変換し、電話番号をハイフンなしの国際形式に正規化しています。normalize=Trueオプションを使用することで、検証時に正規化も行われます。

第9章：ネストされたスキーマの活用

複雑なデータ構造を扱う際、ネストされたスキーマを使用すると便利です。Cerberusは深くネストされたデータ構造も簡単に検証できます。

from cerberus import Validator

schema = {
    'user': {
        'type': 'dict',
        'schema': {
            'name': {'type': 'string'},
            'address': {
                'type': 'dict',
                'schema': {
                    'street': {'type': 'string'},
                    'city': {'type': 'string'},
                    'zip': {'type': 'string', 'regex': '^[0-9]{7}$'}
                }
            },
            'orders': {
                'type': 'list',
                'schema': {
                    'type': 'dict',
                    'schema': {
                        'product': {'type': 'string'},
                        'quantity': {'type': 'integer', 'min': 1},
                        'price': {'type': 'float', 'min': 0}
                    }
                }
            }
        }
    }
}

v = Validator(schema)

data = {
    'user': {
        'name': '田中三郎',
        'address': {
            'street': '桜木町1-2-3',
            'city': '横浜市',
            'zip': '2310062'
        },
        'orders': [
            {'product': 'ノートPC', 'quantity': 1, 'price': 150000.0},
            {'product': 'マウス', 'quantity': 2, 'price': 3000.0}
        ]
    }
}

if v.validate(data):
    print("複雑なネストされたデータ構造も正しく検証されました")
else:
    print("エラー:", v.errors)

この例を使い検証することができます。

第10章：カスタムエラーメッセージの設定

Cerberusのデフォルトエラーメッセージは英語ですが、日本語のアプリケーションでは日本語のエラーメッセージが望ましいでしょう。カスタムエラーメッセージを設定する方法を見てみましょう。

from cerberus import Validator

schema = {
    'username': {
        'type': 'string',
        'minlength': 3,
        'maxlength': 20,
        'regex': '^[a-zA-Z0-9_]+$',
        'error_messages': {
            'minlength': 'ユーザー名は3文字以上である必要があります',
            'maxlength': 'ユーザー名は20文字以下である必要があります',
            'regex': 'ユーザー名は英数字とアンダースコアのみ使用できます'
        }
    },
    'age': {
        'type': 'integer',
        'min': 18,
        'max': 100,
        'error_messages': {
            'min': '18歳以上である必要があります',
            'max': '100歳以下である必要があります'
        }
    }
}

v = Validator(schema)

data1 = {'username': 'a', 'age': 15}
data2 = {'username': 'user@name', 'age': 120}

if not v.validate(data1):
    print("エラー (data1):", v.errors)

if not v.validate(data2):
    print("エラー (data2):", v.errors)

このように、error_messagesキーを使用することで、各フィールドの各ルールに対してカスタムエラーメッセージを設定できます。

第11章：スキーマの再利用と拡張

大規模なアプリケーションでは、スキーマの再利用や拡張が必要になることがあります。Cerberusでは、Pythonの辞書操作を利用してこれを実現できます。

from cerberus import Validator

# 基本的な住所スキーマ
address_schema = {
    'street': {'type': 'string'},
    'city': {'type': 'string'},
    'zip': {'type': 'string', 'regex': '^[0-9]{7}$'}
}

# ユーザースキーマ（住所スキーマを含む）
user_schema = {
    'name': {'type': 'string'},
    'email': {'type': 'string', 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'},
    'address': {'type': 'dict', 'schema': address_schema}
}

# 会社スキーマ（住所スキーマを拡張）
company_schema = {
    'name': {'type': 'string'},
    'address': {
        'type': 'dict',
        'schema': {
            **address_schema,
            'building': {'type': 'string'}
        }
    },
    'employees': {'type': 'list', 'schema': {'type': 'dict', 'schema': user_schema}}
}

v = Validator()

user_data = {
    'name': '山田太郎',
    'email': '[e-mail]',
    'address': {
        'street': '桜木町1-2-3',
        'city': '横浜市',
        'zip': '2310062'
    }
}

company_data = {
    'name': '株式会社サンプル',
    'address': {
        'street': '丸の内1-1-1',
        'city': '東京都',
        'zip': '1000001',
        'building': '東京駅ビル'
    },
    'employees': [user_data]
}

print("ユーザーデータ検証:", v.validate(user_data, user_schema))
print("会社データ検証:", v.validate(company_data, company_schema))

この例では、address_schemaを基本として、user_schemaとcompany_schemaを作成しています。company_schemaではaddress_schemaを拡張してbuildingフィールドを追加しています。

第12章：動的スキーマの生成

時には、実行時に動的にスキーマを生成したい場合があります。例えば、データベースのスキーマに基づいてCerberusのスキーマを生成する場合などです。

from cerberus import Validator

def generate_schema_from_db(table_info):
    schema = {}
    type_mapping = {
        'INTEGER': 'integer',
        'TEXT': 'string',
        'REAL': 'float',
        'BOOLEAN': 'boolean'
    }
    
    for column in table_info:
        name, db_type, required, max_length = column
        field_schema = {'type': type_mapping.get(db_type, 'string')}
        
        if required:
            field_schema['required'] = True
        
        if max_length:
            field_schema['maxlength'] = max_length
        
        schema[name] = field_schema
    
    return schema

# データベースのテーブル情報を模倣
table_info = [
    ('id', 'INTEGER', True, None),
    ('name', 'TEXT', True, 100),
    ('age', 'INTEGER', False, None),
    ('height', 'REAL', False, None),
    ('is_student', 'BOOLEAN', True, None)
]

generated_schema = generate_schema_from_db(table_info)
print("生成されたスキーマ:", generated_schema)

v = Validator(generated_schema)

data = {
    'id': 1,
    'name': '佐藤花子',
    'age': 25,
    'height': 160.5,
    'is_student': False
}

if v.validate(data):
    print("動的に生成されたスキーマでデータが正しく検証されました")
else:
    print("エラー:", v.errors)

この例では、データベースのテーブル情報から動的にCerberusのスキーマを生成しています。実際のアプリケーションでは、データベースの情報を読み取り、それに基づいてスキーマを生成することができます。

第13章：複数のスキーマを使用した検証

時には、同じデータに対して複数の異なるスキーマを適用したい場合があります。例えば、ユーザーの権限レベルに応じて異なる検証ルールを適用する場合などです。

from cerberus import Validator

# 基本的なユーザースキーマ
base_schema = {
    'username': {'type': 'string', 'required': True},
    'email': {'type': 'string', 'required': True, 'regex': '^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'},
    'age': {'type': 'integer', 'min': 0}
}

# 管理者用の拡張スキーマ
admin_schema = {
    **base_schema,
    'role': {'type': 'string', 'allowed': ['admin', 'superadmin'], 'required': True},
    'access_level': {'type': 'integer', 'min': 1, 'max': 10, 'required': True}
}

v = Validator()

def validate_user(data, is_admin=False):
    schema = admin_schema if is_admin else base_schema
    if v.validate(data, schema):
        print("ユーザーデータは有効です")
        return True
    else:
        print("エラー:", v.errors)
        return False

# 一般ユーザーのデータ
user_data = {
    'username': 'user123',
    'email': '[e-mail]',
    'age': 25
}

# 管理者のデータ
admin_data = {
    'username': 'admin456',
    'email': '[e-mail]',
    'age': 30,
    'role': 'admin',
    'access_level': 5
}

print("一般ユーザーの検証:")
validate_user(user_data)

print("\n管理者の検証 (一般ユーザースキーマ):")
validate_user(admin_data)

print("\n管理者の検証 (管理者スキーマ):")
validate_user(admin_data, is_admin=True)

この例では、基本的なユーザースキーマと、それを拡張した管理者用スキーマを定義しています。validate_user関数では、is_adminフラグに基づいて適切なスキーマを選択して検証を行います。

第14章：カスタムバリデータの作成

Cerberusの標準機能だけでは足りない場合、カスタムバリデータを作成することができます。例えば、日本の電話番号を検証する独自のバリデータを作ってみましょう。

import re
from cerberus import Validator

class JapaneseValidator(Validator):
    def _validate_is_japanese_phone(self, is_japanese_phone, field, value):
        """ 日本の電話番号形式を検証します。

        The rule's arguments are validated against this schema:
        {'type': 'boolean'}
        """
        if is_japanese_phone:
            pattern = r'^(0[5-9]0-?[0-9]{4}-?[0-9]{4}|0[1-9][1-9]-?[0-9]{3}-?[0-9]{4})$'
            if not re.match(pattern, value):
                self._error(field, "有効な日本の電話番号ではありません")

schema = {
    'name': {'type': 'string'},
    'phone': {'type': 'string', 'is_japanese_phone': True}
}

v = JapaneseValidator(schema)

data1 = {'name': '山田太郎', 'phone': '090-1234-5678'}
data2 = {'name': '佐藤花子', 'phone': '03-1234-5678'}
data3 = {'name': '鈴木一郎', 'phone': '1234-5678'}

print("データ1の検証:", v.validate(data1))
print("データ2の検証:", v.validate(data2))
print("データ3の検証:", v.validate(data3))
if not v.validate(data3):
    print("エラー:", v.errors)

この例では、JapaneseValidatorクラスを作成し、_validate_is_japanese_phoneメソッドを定義しています。このメソッドは、日本の携帯電話番号と固定電話番号の形式を正規表現で検証します。

第15章：非同期バリデーションの実装

最後に、非同期処理が必要な場合のバリデーション方法を見てみましょう。例えば、データベースやAPIを呼び出して検証を行う必要がある場合などです。

import asyncio
from cerberus import Validator

class AsyncValidator(Validator):
    async def validate_async(self, document, schema=None, update=False, normalize=True):
        self._errors = {}
        self.document = document

        if schema is not None:
            self.schema = schema
        elif self.schema is None:
            raise RuntimeError("Schema is not specified")

        if document is None:
            self._error("document", "document is null")
            return False

        # 非同期で検証を行う
        tasks = []
        for field in self.schema:
            if field in self.document:
                tasks.append(self._validate_field_async(field))
        
        await asyncio.gather(*tasks)

        return len(self._errors) == 0

    async def _validate_field_async(self, field):
        # ここで非同期の検証ロジックを実装
        # 例: データベースやAPIを呼び出して検証
        await asyncio.sleep(1)  # 非同期処理をシミュレート
        self._validate(self.document[field], self.schema[field], field)

async def main():
    schema = {
        'username': {'type': 'string', 'required': True},
        'email': {'type': 'string', 'required': True}
    }

    v = AsyncValidator(schema)

    data = {
        'username': 'user123',
        'email': '[e-mail]'
    }

    is_valid = await v.validate_async(data)
    if is_valid:
        print("データは有効です")
    else:
        print("エラー:", v.errors)

asyncio.run(main())

この例では、AsyncValidatorクラスを作成し、validate_asyncメソッドを実装しています。このメソッドは各フィールドの検証を非同期で行います。実際のアプリケーションでは、_validate_field_asyncメソッド内で、データベースクエリやAPI呼び出しなどの非同期処理を行うことができます。

以上で、Cerberusライブラリの詳細な解説を終わります。このライブラリを使いこなすことで、Pythonアプリケーションのデータ検証をより堅牢かつ柔軟に行うことができます。Cerberusは、シンプルな使用法から複雑な検証ロジックまで幅広くサポートしており、多くのプロジェクトで重宝されるツールです。ぜひ、実際のプロジェクトで活用してみてください。