Pydantic异步校验器深：构建高并发验证系统
title: 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
| 错误信息 | 原因分析 | 解决方案 |
|---|---|---|
| RuntimeError: 事件循环未找到 | 在非异步环境调用校验器 | 使用asyncio.run()封装 |
| ValidationError: 缺少await调用 | 忘记添加await关键字 | 检查所有异步操作的await |
| TimeoutError: 验证超时 | 未设置合理的超时限制 | 增加asyncio.timeout区块 |
| TypeError: 无效的异步生成器 | 错误处理异步流数据 | 使用aiostream库进行流控制 |
评论
发表评论