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
        )