使用 schema 库，自定义较复杂的校验方法

一、前置说明

1、总体目录

《 pyparamvalidate 参数校验器，从编码到发布全过程》

2、相关回顾

添加常用校验方法，校验常见数据格式
添加自定义校验方法，让用户自定义校验规则

3、本节目标

了解 schema 库的基本用法
使用 schema 库，自定义较复杂的校验方法

二、操作步骤

1、项目目录

atme : @me 用于存放临时的代码片断或其它内容。
pyparamvalidate : 新建一个与项目名称同名的package，为了方便发布至 pypi 。
core : 用于存放核心代码。
tests : 用于存放测试代码。
utils : 用于存放一些工具类或方法。

2、依赖包安装及说明

pip install schema

schema 是一个轻量级的库，用于数据验证和结构定义。它支持定义嵌套的结构，并且可以进行复杂的验证。
schema GitHub 仓库地址：https://github.com/keleshev/schema

3、代码实现

demo


...

class Validator(metaclass=RaiseExceptionMeta):
    def __init__(self, value, field=None, rule_des=None):
        self.value = value
        self._field = field
        self._rule_des = rule_des

    def schema_validate(self, schema: Schema) -> Self:
        """
        schema 官方参考文档： https://pypi.org/project/schema/

        下面是涵盖了 Schema 大部分特性的示例说明:

        1. 定义 schema

            # 自定义处理函数，首字母大写
            def capitalize(value):
                return value.capitalize()


            # 邮箱格式验证函数
            def validate_email(value):
                email_regex = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
                return bool(re.match(email_regex, value))


            user_schema = schema.Schema({
                'username': schema.And(str, lambda s: len(s.strip()) > 0, error='Username cannot be empty or contain only spaces'),
                'phone_number': schema.Regex(r'^\d{11}$', error='Invalid phone number format. It should be a 10-digit number.'),
                'email': schema.And(schema.Or(str, None), lambda s: validate_email(s) if s is not None else True, error='Invalid email format'),
                'age': schema.And(int, lambda n: 0 <= n <= 120, error='Age must be an integer between 0 and 120'),
                'gender': schema.And(str, lambda s: s.lower() in ['male', 'female', 'other'], error='Invalid gender'),
                'family_members': schema.And(schema.Use(list), [schema.Use(capitalize)]),
                'others': {
                    'address': schema.And(str, lambda s: s.strip(), error='Address must be a non-empty string'),
                    'blog': schema.Or(None, schema.Regex(r'^https?://\S+$', error='Invalid blog format. It should be a valid URL starting with http:// or https://')),
                    'other': schema.Or(str, None)
                }
            })

        2. 使用 schema 进行校验

            valid_data = {
            'username': 'JohnDoe',
            'phone_number': '13888886666',
            'email': 'john@example.com',
            'age': 25,
            'gender': 'male',
            'family_members': ['Alice', 'Bob', 'Charlie'],
            'others': {
                'address': '123 Main St',
                'blog': 'http://example.com',
                'other': 'Some additional info'
                }
            }

            Validator(valid_data).schema_validate(user_schema)

        """
        if not isinstance(schema, Schema):
            raise CallValidateMethodError(f'{schema} must be a instance of Schema, not {type(schema)}, '
                                          f'Please use "schema.Schema()" to initialize the schema.'
                                          f'Documentation: https://pypi.org/project/schema/')

        # 将 validate 之后的值赋值给 self.value，因为 schema 在校验的过程中可以对 value 进行预处理
        self.value = schema.validate(self.value)
        return self

	...

3、测试代码

pyparamvalidate/tests/test_validator.py


import os
import re

import pytest
import schema

from pyparamvalidate.core.validator import Validator, CallValidateMethodError


# 自定义处理函数，首字母大写
def capitalize(value):
    return value.capitalize()


# 邮箱格式验证函数
def validate_email(value):
    email_regex = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
    return bool(re.match(email_regex, value))


reference_correct_data = {
    'username': 'JohnDoe',
    'phone_number': '13888886666',
    'email': 'john@example.com',
    'age': 25,
    'gender': 'male',
    'family_members': ['Alice', 'Bob', 'Charlie'],
    'others': {
        'address': '123 Main St',
        'blog': 'http://example.com',
        'other': 'Some additional info'
    }
}

user_schema = schema.Schema({
    'username': schema.And(str, lambda s: len(s.strip()) > 0, error='Username cannot be empty or contain only spaces'),
    'phone_number': schema.Regex(r'^\d{11}$', error='Invalid phone number format. It should be a 10-digit number.'),
    'email': schema.And(schema.Or(str, None), lambda s: validate_email(s) if s else True, error='Invalid email format'),
    'age': schema.And(int, lambda n: 0 <= n <= 120, error='Age must be an integer between 0 and 120'),
    'gender': schema.And(str, lambda s: s.lower() in ['male', 'female', 'other'], error='Invalid gender'),
    'family_members': schema.And(schema.Use(list), [schema.Use(capitalize)]),
    'others': {
        'address': schema.And(str, lambda s: s.strip(), error='Address must be a non-empty string'),
        'blog': schema.Or(None, schema.Regex(r'^https?://\S+$', error='Invalid blog format. It should be a valid URL starting with http:// or https://')),
        'other': schema.Or(str, None)
    }
})


def test_schema_validate_01():
    valid_data = {
        'username': 'JohnDoe',
        'phone_number': '13888886666',
        'email': 'john@example.com',
        'age': 25,
        'gender': 'male',
        'family_members': ['Alice', 'Bob', 'Charlie'],
        'others': {
            'address': '123 Main St',
            'blog': 'http://example.com',
            'other': 'Some additional info'
        }
    }
    assert Validator(valid_data).schema_validate(user_schema).value == valid_data


# 反向测试用例
def test_schema_validate_02():
    invalid_data_list = [
        # 无效的用户名（空字符串）
        {'username': '   ', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的邮箱格式
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'invalidemail', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的年龄（负数）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': -5, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的性别
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25,
         'gender': 'unknown',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的博客格式（不是以 http:// 或 https:// 开头）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'], 'others': {'address': '123 Main St', 'blog': 'invalidblog',
                                                                   'other': 'Some additional info'}},
        # 无效的家庭成员列表（包含空字符串）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', '', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的地址（空字符串）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'], 'others': {'address': '', 'other': 'Some additional info'}},
        # 无效的电话号码格式（不是10位数字）
        {'username': 'JohnDoe', 'phone_number': '12345', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'other': 'Some additional info'}},
        # 无效的博客格式（None 但不为空字符串）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'],
         'others': {'address': '123 Main St', 'blog': '', 'other': 'Some additional info'}},
        # 无效的博客格式（不是以 http:// 或 https:// 开头）
        {'username': 'JohnDoe', 'phone_number': '13888886666', 'email': 'john@example.com', 'age': 25, 'gender': 'male',
         'family_members': ['Alice', 'Bob', 'Charlie'], 'others': {'address': '123 Main St', 'blog': 'invalidblog',
                                                                   'other': 'Some additional info'}}
    ]

    for invalid_data in invalid_data_list:
        with pytest.raises(schema.SchemaError) as exc_info:
            Validator(invalid_data).schema_validate(user_schema)
        print(exc_info.value)
        print('===============================')

4、日志输出

执行 test 的日志如下，验证通过：

============================= test session starts =============================
collecting ... collected 29 items

test_validator.py::test_schema_validate_01 PASSED                        [  3%]
test_validator.py::test_schema_validate_02 PASSED                        [  6%]Username cannot be empty or contain only spaces
===============================
Invalid email format
===============================
Age must be an integer between 0 and 120
===============================
Invalid gender
===============================
Invalid blog format. It should be a valid URL starting with http:// or https://
===============================
Key 'others' error:
Missing key: 'blog'
===============================
Address must be a non-empty string
===============================
Invalid phone number format. It should be a 10-digit number.
===============================
Invalid blog format. It should be a valid URL starting with http:// or https://
===============================
Invalid blog format. It should be a valid URL starting with http:// or https://
===============================