Pydantic异步校验器深：构建高并发验证系统

from pydantic import BaseModel, validator
import asyncio


class AsyncValidator(BaseModel):
    domain: str

    @validator("domain", pre=True)
    async def check_dns_record(cls, v):
        reader, writer = await asyncio.open_connection("8.8.8.8", 53)
        # 发送DNS查询请求（示例代码）
        writer.write(b"DNS query packet")
        await writer.drain()
        response = await reader.read(1024)
        writer.close()
        return v if b"valid" in response else "invalid_domain"

import aiohttp


class BatchAPIValidator(BaseModel):
    endpoints: list[str]

    @validator("endpoints")
    async def validate_apis(cls, v):
        async with aiohttp.ClientSession() as session:
            tasks = [session.head(url) for url in v]
            responses = await asyncio.gather(*tasks)
            return [
                url for url, resp in zip(v, responses)
                if resp.status < 400
            ]

from sqlalchemy.ext.asyncio import AsyncSession


class UserValidator(BaseModel):
    username: str

    @validator("username")
    async def check_unique(cls, v):
        async with AsyncSession(engine) as session:
            result = await session.execute(
                select(User).where(User.username == v)
            )
            if result.scalars().first():
                raise ValueError("用户名已存在")
            return v

from redis.asyncio import Redis


class OrderValidator(BaseModel):
    order_id: str

    @validator("order_id")
    async def check_distributed_lock(cls, v):
        redis = Redis.from_url("redis://localhost")
        async with redis.lock(f"order_lock:{v}", timeout=10):
            if await redis.exists(f"order:{v}"):
                raise ValueError("订单重复提交")
            await redis.setex(f"order:{v}", 300, "processing")
            return v

from abc import ABC, abstractmethod


class AsyncValidationStrategy(ABC):
    @abstractmethod
    async def validate(self, value): ...


class EmailStrategy(AsyncValidationStrategy):
    async def validate(self, value):
        await asyncio.sleep(0.1)  # 模拟DNS查询
        return "@" in value


class AsyncCompositeValidator(BaseModel):
    email: str
    strategy: AsyncValidationStrategy

    @validator("email")
    async def validate_email(cls, v, values):
        if not await values["strategy"].validate(v):
            raise ValueError("邮箱格式错误")
        return v

import aiostream


class StreamValidator(BaseModel):
    data_stream: AsyncGenerator

    @validator("data_stream")
    async def process_stream(cls, v):
        async with aiostream.stream.iterate(v) as stream:
            return await (
                stream
                .map(lambda x: x * 2)
                .filter(lambda x: x < 100)
                .throttle(10)  # 限流10条/秒
                .list()
            )

class PaymentValidator(BaseModel):
    user_id: int
    balance: float = None

    @validator("user_id")
    async def fetch_balance(cls, v):
        async with aiohttp.ClientSession() as session:
            async with session.get(f"/users/{v}/balance") as resp:
                return await resp.json()

    @validator("balance", always=True)
    async def check_sufficient(cls, v):
        if v < 100:
            raise ValueError("余额不足最低限额")

class TimeoutValidator(BaseModel):
    api_url: str

    @validator("api_url")
    async def validate_with_timeout(cls, v):
        try:
            async with asyncio.timeout(5):
                async with aiohttp.ClientSession() as session:
                    async with session.get(v) as resp:
                        return v if resp.status == 200 else "invalid"
        except TimeoutError:
            raise ValueError("API响应超时")

from pydantic import ValidationError


class BulkValidator(BaseModel):
    items: list[str]

    @validator("items")
    async def bulk_check(cls, v):
        errors = []
        for item in v:
            try:
                await external_api.check(item)
            except Exception as e:
                errors.append(f"{item}: {str(e)}")
        if errors:
            raise ValueError("\n".join(errors))
        return v