IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python校验封装案例如何封装参数校验

wen python案例 32

本文目录导读：

Python校验封装案例如何封装参数校验

  1. 基础装饰器方式
  2. dataclass + 验证器
  3. Pydantic模型验证
  4. 链式校验器
  5. 综合验证框架
  6. 性能优化版

我来详细介绍Python中参数校验的封装方法,包含多种实践方案。

基础装饰器方式

from functools import wraps
from typing import Any, Callable
def validate_args(**validators):
    """
    参数校验装饰器
    validators: 字段名 -> 校验函数 的映射
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 获取函数参数名
            import inspect
            sig = inspect.signature(func)
            bound_args = sig.bind(*args, **kwargs)
            bound_args.apply_defaults()
            # 校验参数
            for param_name, validator in validators.items():
                if param_name in bound_args.arguments:
                    value = bound_args.arguments[param_name]
                    if not validator(value):
                        raise ValueError(
                            f"参数 '{param_name}' 校验失败，值为: {value}"
                        )
            return func(*args, **kwargs)
        return wrapper
    return decorator
# 使用示例
def is_positive(value: Any) -> bool:
    return isinstance(value, (int, float)) and value > 0
def is_string(value: Any) -> bool:
    return isinstance(value, str) and len(value) > 0
@validate_args(
    age=is_positive,
    name=is_string
)
def create_user(name: str, age: int):
    return f"User {name}, Age {age}"
# 测试
try:
    print(create_user("Alice", 25))  # 正常
    print(create_user("", -5))       # 报错
except ValueError as e:
    print(f"错误: {e}")

dataclass + 验证器

from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
class ValidationError(Exception):
    pass
@dataclass
class UserModel:
    """用户数据模型"""
    username: str
    email: str
    age: int
    phone: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    def __post_init__(self):
        self.validate()
    def validate(self):
        """验证所有字段"""
        errors = []
        # 验证用户名
        if not self.username or len(self.username) < 3:
            errors.append("用户名至少3个字符")
        if not self.username.isalnum():
            errors.append("用户名只能包含字母和数字")
        # 验证邮箱
        if not self.email or '@' not in self.email:
            errors.append("邮箱格式不正确")
        # 验证年龄
        if not isinstance(self.age, int) or self.age < 0 or self.age > 150:
            errors.append("年龄必须在0-150之间")
        # 验证手机号
        if self.phone:
            import re
            if not re.match(r'^1[3-9]\d{9}$', self.phone):
                errors.append("手机号格式不正确")
        if errors:
            raise ValidationError("; ".join(errors))
# 使用示例
try:
    user = UserModel("alice", "alice@example.com", 25)
    print(f"创建用户成功: {user}")
    # 错误示例
    bad_user = UserModel("ab", "invalid-email", -5)
except ValidationError as e:
    print(f"验证失败: {e}")

Pydantic模型验证

from pydantic import BaseModel, validator, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"
class UserCreate(BaseModel):
    """用户创建请求模型"""
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: str = Field(..., description="邮箱地址")
    age: int = Field(..., ge=0, le=150, description="年龄")
    role: UserRole = Field(default=UserRole.USER, description="角色")
    phone: Optional[str] = Field(None, description="手机号")
    tags: List[str] = Field(default_factory=list, description="标签")
    @validator('username')
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        return v
    @validator('email')
    def email_valid(cls, v):
        if '@' not in v:
            raise ValueError('邮箱格式不正确')
        return v
    @validator('phone')
    def phone_valid(cls, v):
        if v is not None:
            import re
            if not re.match(r'^1[3-9]\d{9}$', v):
                raise ValueError('手机号格式不正确')
        return v
    @validator('tags')
    def tags_unique(cls, v):
        if len(v) != len(set(v)):
            raise ValueError('标签不能重复')
        return v
class UserResponse(BaseModel):
    """用户响应模型"""
    id: int
    username: str
    email: str
    role: UserRole
    created_at: datetime
    class Config:
        orm_mode = True
# 使用示例
try:
    user_data = {
        "username": "alice123",
        "email": "alice@example.com",
        "age": 25,
        "tags": ["python", "developer"]
    }
    user = UserCreate(**user_data)
    print(f"创建成功: {user.dict()}")
    # 错误示例
    bad_data = {
        "username": "ab",
        "email": "invalid",
        "age": 200
    }
    bad_user = UserCreate(**bad_data)
except Exception as e:
    print(f"验证失败: {e}")

链式校验器

from typing import Any, Callable, List, Tuple, Union
from functools import reduce
class ValidatorChain:
    """链式校验器"""
    def __init__(self, value: Any, name: str = ""):
        self.value = value
        self.name = name
        self.errors: List[str] = []
        self._validators: List[Tuple[Callable, str]] = []
    def required(self, message: str = None) -> 'ValidatorChain':
        """非空验证"""
        def _required(v):
            return v is not None and str(v).strip() != ""
        msg = message or f"{self.name} 不能为空"
        self._validators.append((_required, msg))
        return self
    def type_check(self, expected_type: type, message: str = None) -> 'ValidatorChain':
        """类型验证"""
        def _type_check(v):
            return isinstance(v, expected_type)
        msg = message or f"{self.name} 必须是 {expected_type.__name__} 类型"
        self._validators.append((_type_check, msg))
        return self
    def min_length(self, min_len: int, message: str = None) -> 'ValidatorChain':
        """最小长度验证"""
        def _min_length(v):
            return len(str(v)) >= min_len
        msg = message or f"{self.name} 长度不能小于 {min_len}"
        self._validators.append((_min_length, msg))
        return self
    def max_length(self, max_len: int, message: str = None) -> 'ValidatorChain':
        """最大长度验证"""
        def _max_length(v):
            return len(str(v)) <= max_len
        msg = message or f"{self.name} 长度不能大于 {max_len}"
        self._validators.append((_max_length, msg))
        return self
    def regex(self, pattern: str, message: str = None) -> 'ValidatorChain':
        """正则表达式验证"""
        import re
        def _regex(v):
            return bool(re.match(pattern, str(v)))
        msg = message or f"{self.name} 格式不正确"
        self._validators.append((_regex, msg))
        return self
    def custom(self, validator: Callable, message: str) -> 'ValidatorChain':
        """自定义验证"""
        self._validators.append((validator, message))
        return self
    def validate(self) -> Union[Any, None]:
        """执行验证"""
        for validator, message in self._validators:
            if not validator(self.value):
                self.errors.append(message)
        if self.errors:
            raise ValueError("; ".join(self.errors))
        return self.value
# 高级封装
class ValidatedField:
    """验证字段描述符"""
    def __init__(self, validator_func: Callable):
        self.validator_func = validator_func
        self.name = ""
    def __set_name__(self, owner, name):
        self.name = name
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return obj.__dict__.get(self.name)
    def __set__(self, obj, value):
        validated_value = self.validator_func(value, self.name)
        obj.__dict__[self.name] = validated_value
class ValidatedUser:
    """使用验证字段的类"""
    username = ValidatedField(
        lambda v, n: ValidatorChain(v, n)
            .required()
            .type_check(str)
            .min_length(3, "用户名至少3个字符")
            .max_length(50, "用户名最多50个字符")
            .custom(lambda x: x.isalnum(), "用户名只能包含字母和数字")
            .validate()
    )
    age = ValidatedField(
        lambda v, n: ValidatorChain(v, n)
            .required()
            .type_check(int)
            .custom(lambda x: 0 <= x <= 150, "年龄必须在0-150之间")
            .validate()
    )
    email = ValidatedField(
        lambda v, n: ValidatorChain(v, n)
            .required()
            .type_check(str)
            .regex(r'^[\w\.-]+@[\w\.-]+\.\w+$', "邮箱格式不正确")
            .validate()
    )
    def __init__(self, username: str, age: int, email: str):
        self.username = username
        self.age = age
        self.email = email
    def __repr__(self):
        return f"ValidatedUser(username='{self.username}', age={self.age}, email='{self.email}')"
# 使用示例
try:
    user = ValidatedUser("alice123", 25, "alice@example.com")
    print(f"创建成功: {user}")
    # 错误示例
    bad_user = ValidatedUser("ab", -5, "invalid")
except ValueError as e:
    print(f"验证失败: {e}")

综合验证框架

from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field
import re
from datetime import datetime
class ValidationRule:
    """验证规则基类"""
    def validate(self, value: Any, field_name: str) -> None:
        raise NotImplementedError
    def get_message(self, field_name: str) -> str:
        raise NotImplementedError
class Required(ValidationRule):
    def validate(self, value: Any, field_name: str) -> None:
        if value is None or str(value).strip() == "":
            raise ValueError(f"{field_name} 不能为空")
class StringType(ValidationRule):
    def validate(self, value: Any, field_name: str) -> None:
        if not isinstance(value, str):
            raise ValueError(f"{field_name} 必须是字符串类型")
class IntType(ValidationRule):
    def validate(self, value: Any, field_name: str) -> None:
        if not isinstance(value, int):
            raise ValueError(f"{field_name} 必须是整数类型")
class MinLength(ValidationRule):
    def __init__(self, min_len: int):
        self.min_len = min_len
    def validate(self, value: Any, field_name: str) -> None:
        if len(str(value)) < self.min_len:
            raise ValueError(f"{field_name} 长度不能小于 {self.min_len}")
class MaxLength(ValidationRule):
    def __init__(self, max_len: int):
        self.max_len = max_len
    def validate(self, value: Any, field_name: str) -> None:
        if len(str(value)) > self.max_len:
            raise ValueError(f"{field_name} 长度不能大于 {self.max_len}")
class Regex(ValidationRule):
    def __init__(self, pattern: str, message: str = None):
        self.pattern = pattern
        self.message = message
    def validate(self, value: Any, field_name: str) -> None:
        if not re.match(self.pattern, str(value)):
            raise ValueError(self.message or f"{field_name} 格式不正确")
class Range(ValidationRule):
    def __init__(self, min_val: Any = None, max_val: Any = None):
        self.min_val = min_val
        self.max_val = max_val
    def validate(self, value: Any, field_name: str) -> None:
        if self.min_val is not None and value < self.min_val:
            raise ValueError(f"{field_name} 不能小于 {self.min_val}")
        if self.max_val is not None and value > self.max_val:
            raise ValueError(f"{field_name} 不能大于 {self.max_val}")
class FieldValidator:
    """字段验证器"""
    def __init__(self, rules: List[ValidationRule]):
        self.rules = rules
    def validate(self, value: Any, field_name: str) -> None:
        for rule in self.rules:
            rule.validate(value, field_name)
class ModelValidator:
    """模型验证器"""
    def __init__(self):
        self._fields: Dict[str, FieldValidator] = {}
    def add_field(self, name: str, rules: List[ValidationRule]) -> 'ModelValidator':
        self._fields[name] = FieldValidator(rules)
        return self
    def validate(self, data: Dict[str, Any]) -> Dict[str, Any]:
        errors = {}
        for field_name, validator in self._fields.items():
            value = data.get(field_name)
            try:
                validator.validate(value, field_name)
            except ValueError as e:
                errors[field_name] = str(e)
        if errors:
            raise ValidationError(errors)
        return data
    def create_validator(self) -> Callable:
        """创建可调用的验证器"""
        def validator(data: Dict[str, Any]) -> Dict[str, Any]:
            return self.validate(data)
        return validator
# 使用示例
user_validator = ModelValidator()
user_validator.add_field("username", [
    Required(),
    StringType(),
    MinLength(3),
    MaxLength(50),
    Regex(r'^\w+$', "用户名只能包含字母、数字和下划线")
])
user_validator.add_field("age", [
    Required(),
    IntType(),
    Range(min_val=0, max_val=150)
])
user_validator.add_field("email", [
    Required(),
    StringType(),
    Regex(r'^[\w\.-]+@[\w\.-]+\.\w+$', "邮箱格式不正确")
])
# 测试
test_data = {
    "username": "alice_123",
    "age": 25,
    "email": "alice@example.com"
}
try:
    validated_data = user_validator.validate(test_data)
    print(f"验证通过: {validated_data}")
except ValidationError as e:
    print(f"验证失败: {e.errors}")

性能优化版

from functools import lru_cache
from typing import Any, Dict, List, Callable
import time
class CachedValidator:
    """带缓存的验证器"""
    def __init__(self):
        self._validators: Dict[str, Callable] = {}
    def add_rule(self, field: str, validator: Callable):
        self._validators[field] = validator
    @lru_cache(maxsize=1000)
    def _validate_field(self, field: str, value_hash: int) -> bool:
        """缓存验证结果"""
        if field not in self._validators:
            return True
        validator = self._validators[field]
        # 从hash恢复实际值（示例简化）
        return True
    def validate(self, data: Dict[str, Any]) -> bool:
        for field, value in data.items():
            value_hash = hash(str(value))
            if not self._validate_field(field, value_hash):
                return False
        return True
# 性能测试
class PerformanceTest:
    def __init__(self):
        self.validator = CachedValidator()
        self.validator.add_rule("username", lambda x: len(x) >= 3)
        self.validator.add_rule("age", lambda x: 0 <= x <= 150)
    def test_performance(self, iterations: int = 10000):
        test_data = {"username": "alice", "age": 25}
        start = time.time()
        for _ in range(iterations):
            self.validator.validate(test_data)
        end = time.time()
        print(f"执行 {iterations} 次验证耗时: {end - start:.4f} 秒")
# 运行性能测试
if __name__ == "__main__":
    perf_test = PerformanceTest()
    perf_test.test_performance()

选择建议

  1. 简单项目: 使用装饰器方式（方案1）
  2. 数据处理类: 使用dataclass验证（方案2）
  3. API开发: 使用Pydantic（方案3）
  4. 复杂验证逻辑: 使用链式验证器（方案4）
  5. 框架开发: 使用综合验证框架（方案5）
  6. 高性能需求: 使用缓存验证器（方案6）

最佳实践

  • 验证逻辑应集中管理
  • 错误信息要清晰明确
  • 考虑性能影响
  • 支持自定义验证规则
  • 保持验证器可测试性

这些方案可以根据项目的具体需求进行选择和组合。

上一篇Python定时封装案例如何封装定时任务

下一篇Python返回封装案例如何统一返回格式

相关文章

抱歉，评论功能暂时关闭!