Pydantic字段级校验：解锁@validator的12种应用
title: Pydantic字段级校验：解锁@validator的12种应用
from pydantic import BaseModel, validator
class CurrencyConverter(BaseModel):
amount: str
@validator("amount", pre=True)
def string_to_float(cls, v):
return float(v.strip("$"))
# 自动转换 "$100.5" → 100.5
print(CurrencyConverter(amount="$100.5").amount)
import re
class IdentityForm(BaseModel):
passport: str
@validator("passport")
def validate_passport(cls, v):
if not re.match(r"^[A-PR-WY][1-9]\d\s?\d{4}[A-Z]$", v):
raise ValueError("护照号码格式错误")
return v.upper().replace(" ", "")
from enum import Enum
class Department(Enum):
HR = 1
IT = 2
class Employee(BaseModel):
dept: int
@validator("dept")
def check_department(cls, v):
return Department(v).name # 自动转换数字为枚举名称
class OrderForm(BaseModel):
product_type: str
weight: float
@validator("weight")
def check_weight(cls, v, values):
if values.get("product_type") == "fragile" and v > 10:
raise ValueError("易碎品不得超过10kg")
return v
import os
class EnvAwareValidator(BaseModel):
api_key: str
@validator("api_key")
def check_key_format(cls, v):
env = os.getenv("APP_ENV", "dev")
if env == "prod" and len(v) < 32:
raise ValueError("生产环境密钥强度不足")
return v
class QuerySafe(BaseModel):
search_term: str
@validator("search_term")
def sanitize_input(cls, v):
forbidden = ["'", ";", "--", "/*"]
if any(c in v for c in forbidden):
raise ValueError("检测到危险字符")
return v.replace("%", "\\%")
from html import escape
class CommentForm(BaseModel):
content: str
@validator("content")
def sanitize_html(cls, v):
return escape(v).replace("\n", "<br>")
class AddressNormalizer(BaseModel):
street: str
@validator("street")
def standardize_address(cls, v):
replacements = {
"St.": "Street",
"Ave": "Avenue"
}
for k, v in replacements.items():
v = v.replace(k, v)
return v.title()
from cryptography.fernet import Fernet
class SecureData(BaseModel):
secret: str
@validator("secret")
def encrypt_value(cls, v):
key = Fernet.generate_key()
return Fernet(key).encrypt(v.encode())
import snowflake
class SnowflakeValidator(BaseModel):
object_id: str
@validator("object_id")
def validate_snowflake(cls, v):
try:
snowflake.deconstruct(v)
return v
except Exception:
raise ValueError("非法分布式ID格式")
from decimal import Decimal, ROUND_HALF_UP
class FinancialModel(BaseModel):
amount: float
@validator("amount")
def monetary_precision(cls, v):
return Decimal(str(v)).quantize(
Decimal("0.00"),
rounding=ROUND_HALF_UP
)
| 错误信息 | 原因分析 | 解决方案 |
|---|---|---|
| ValidationError: value is not a valid integer | 类型转换前未清洗数据 | 添加pre=True验证器 |
| ValueError: 检测到危险字符 | SQL注入防御生效 | 使用参数化查询替代直接拼接 |
| AssertionError: 校验顺序错误 | 依赖字段未优先验证 | 调整字段定义顺序 |
| TypeError: 校验器返回类型错误 | 验证器返回值与声明类型不符 | 检查验证器逻辑 |
评论
发表评论