
上周一个老项目原本2000多行的代码,用了一个库后直接缩减到800行。更惊讶的是,不仅代码量减少了,可读性和性能还提升了。这个让我相见恨晚的库就是——pydantic。
今天就来分享,pydantic如何帮我解决那些让Python开发者头疼的问题,以及为什么它应该成为你的标准库之一。
以前写API接口,参数验证是最繁琐的部分。每个接口都要写一堆验证逻辑,既重复又容易出错。
改造前(50+行代码):
fromflaskimportrequest, jsonify
importre
fromdatetimeimportdatetime
@app.route('/api/user', methods=['POST'])
defcreate_user():
data = request.get_json()
# 验证必填字段
if'name'notindata:
returnjsonify({'error': '姓名必填'}), 400
if'email'notindata:
returnjsonify({'error': '邮箱必填'}), 400
if'age'notindata:
returnjsonify({'error': '年龄必填'}), 400
# 验证姓名长度
name = data['name']
iflen(name) <2orlen(name) >20:
returnjsonify({'error': '姓名长度2-20字符'}), 400
# 验证邮箱格式
email = data['email']
ifnotre.match(r'^[\w\.-]+@[\w\.-]+\.\w+$', email):
returnjsonify({'error': '邮箱格式不正确'}), 400
# 验证年龄范围
age = data['age']
ifnotisinstance(age, int) orage<0orage>150:
returnjsonify({'error': '年龄必须在0-150之间'}), 400
# 验证生日(如果有)
if'birthday'indata:
try:
birthday = datetime.strptime(data['birthday'], '%Y-%m-%d')
ifbirthday>datetime.now():
returnjsonify({'error': '生日不能是未来时间'}), 400
exceptValueError:
returnjsonify({'error': '生日格式错误'}), 400
# 所有验证通过,处理业务逻辑
# ... 50行业务代码 ...
returnjsonify({'success': True}), 201改造后(5行代码搞定验证):
frompydanticimportBaseModel, validator, Field
fromdatetimeimportdate
fromtypingimportOptional
classUserCreate(BaseModel):
name: str = Field(..., min_length=2, max_length=20, description="用户姓名")
email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: int = Field(..., ge=0, le=150, description="用户年龄")
birthday: Optional[date] = None
@validator('birthday')
defvalidate_birthday(cls, v):
ifvandv>date.today():
raiseValueError('生日不能是未来时间')
returnv
@app.route('/api/user', methods=['POST'])
defcreate_user():
# 一行代码完成所有验证
user_data = UserCreate(**request.get_json())
# 直接使用验证过的数据
# ... 业务逻辑 ...
returnjsonify({'success': True}), 201效果对比:
以前项目配置散落在各处:环境变量、配置文件、代码中的默认值,管理起来一团糟。
改造前(混乱的配置管理):
# config.py - 混乱的配置管理
importos
importjson
# 尝试从环境变量读取,没有就用默认值
DATABASE_HOST = os.getenv('DB_HOST', 'localhost')
DATABASE_PORT = int(os.getenv('DB_PORT', '3306'))
DATABASE_USER = os.getenv('DB_USER', 'root')
DATABASE_PASSWORD = os.getenv('DB_PASSWORD', '')
# 从配置文件读取
try:
withopen('config.json', 'r') asf:
config_json = json.load(f)
REDIS_HOST = config_json.get('redis_host', 'localhost')
REDIS_PORT = config_json.get('redis_port', 6379)
exceptFileNotFoundError:
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
# 代码中的硬编码默认值
DEBUG = Trueifos.getenv('ENV') == 'development'elseFalse
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
# 使用时需要到处import
# from config import DATABASE_HOST, DATABASE_PORT, ...改造后(统一的配置管理):
# config.py - 优雅的配置管理
frompydanticimportBaseSettings, Field
fromtypingimportOptional
classSettings(BaseSettings):
# 自动从环境变量读取,支持.env文件
database_host: str = Field('localhost', env='DB_HOST')
database_port: int = Field(3306, env='DB_PORT')
database_user: str = Field('root', env='DB_USER')
database_password: str = Field('', env='DB_PASSWORD')
redis_host: str = Field('localhost', env='REDIS_HOST')
redis_port: int = Field(6379, env='REDIS_PORT')
debug: bool = Field(False, env='DEBUG')
log_level: str = Field('INFO', env='LOG_LEVEL')
# 复杂验证
@validator('database_port', 'redis_port')
defvalidate_port(cls, v):
ifnot1<= v<= 65535:
raiseValueError('端口必须在1-65535之间')
returnv
classConfig:
env_file = '.env' # 支持.env文件
env_file_encoding = 'utf-8'
case_sensitive = False # 不区分大小写
# 全局配置实例
settings = Settings()
# 使用方式
print(settings.database_host) # 自动从环境变量或.env读取
print(settings.redis_port) # 类型安全,自动转换优势:
settings对象访问在微服务架构中,经常需要在不同层之间传递数据。以前要定义多个相似的数据类,现在一个模型搞定。
改造前(重复定义数据类):
# 数据库模型
classUserDB:
def__init__(self, id, name, email, age, created_at):
self.id = id
self.name = name
self.email = email
self.age = age
self.created_at = created_at
# API请求模型
classUserCreateRequest:
def__init__(self, name, email, age):
self.name = name
self.email = email
self.age = age
# API响应模型
classUserResponse:
def__init__(self, id, name, email, age, created_at):
self.id = id
self.name = name
self.email = email
self.age = age
self.created_at = created_at
# 业务逻辑中各种转换
defcreate_user(request: UserCreateRequest) ->UserResponse:
# 验证逻辑...
# 转换逻辑...
db_user = UserDB(
id=generate_id(),
name=request.name,
email=request.email,
age=request.age,
created_at=datetime.now()
)
# 保存到数据库...
returnUserResponse(
id=db_user.id,
name=db_user.name,
email=db_user.email,
age=db_user.age,
created_at=db_user.created_at
)改造后(一个模型,多种用途):
frompydanticimportBaseModel, Field
fromdatetimeimportdatetime
fromtypingimportOptional
classUserBase(BaseModel):
"""基础字段"""
name: str = Field(..., min_length=2, max_length=50)
email: str
age: int = Field(..., ge=0, le=150)
classUserCreate(UserBase):
"""创建用户时的字段(不需要id和created_at)"""
pass
classUserDB(UserBase):
"""数据库模型(包含所有字段)"""
id: int
created_at: datetime
classConfig:
orm_mode = True # 支持从ORM对象转换
classUserResponse(UserBase):
"""API响应(可以排除敏感字段)"""
id: int
created_at: datetime
classConfig:
exclude = {'email'} # 响应中排除邮箱
# 使用示例
defcreate_user(user_data: UserCreate) ->UserResponse:
# 自动验证user_data
db_user = UserDB(
id=generate_id(),
**user_data.dict(), # 复用字段
created_at=datetime.now()
)
# 保存数据库...
returnUserResponse.from_orm(db_user) # 自动转换继承链的好处:
UserBase(基础字段)
├── UserCreate(API请求)
├── UserDB(数据库模型)
└── UserResponse(API响应)以前写复杂的业务验证要写很多if-else,现在用pydantic的validator装饰器,逻辑清晰又简洁。
复杂验证示例:
frompydanticimportBaseModel, validator
fromdatetimeimportdate
fromtypingimportList
classOrder(BaseModel):
items: List[str]
quantities: List[int]
order_date: date
delivery_date: date
@validator('items')
defvalidate_items(cls, v):
ifnotv:
raiseValueError('订单不能为空')
iflen(v) >10:
raiseValueError('单次最多购买10种商品')
returnv
@validator('quantities')
defvalidate_quantities(cls, v, values):
if'items'invaluesandlen(v) != len(values['items']):
raiseValueError('商品数量和种类数不匹配')
ifany(q<= 0forqinv):
raiseValueError('商品数量必须大于0')
returnv
@validator('delivery_date')
defvalidate_delivery_date(cls, v, values):
if'order_date'invaluesandv<= values['order_date']:
raiseValueError('配送日期必须在订单日期之后')
if (v-date.today()).days>30:
raiseValueError('配送日期不能超过30天')
returnv
# 跨字段验证
@validator('*')
defvalidate_total_quantity(cls, v, values):
if'quantities'invalues:
total = sum(values['quantities'])
iftotal>100:
raiseValueError('单次订单总数量不能超过100')
returnv如果你用FastAPI开发Web应用,pydantic简直是绝配。自动生成API文档,自动验证请求,开发效率提升不止一倍。
fromfastapiimportFastAPI
frompydanticimportBaseModel
app = FastAPI()
classItem(BaseModel):
name: str
price: float
tags: list[str] = []
@app.post("/items/")
asyncdefcreate_item(item: Item):
# item已经自动验证
return {"item_name": item.name, "item_price": item.price}
@app.put("/items/{item_id}")
asyncdefupdate_item(item_id: int, item: Item, q: str = None):
return {"item_id": item_id, **item.dict(), "q": q}自动生成的API文档:
/docs 查看交互式文档/redoc 查看ReDoc文档1. 动态模型创建:
frompydanticimportcreate_model
# 根据配置动态创建模型
DynamicModel = create_model(
'DynamicModel',
name=(str, ...),
age=(int, 0), # 默认值
__config__={'extra': 'forbid'} # 禁止额外字段
)
# 使用
model = DynamicModel(name='张三')2. 自定义类型:
frompydanticimportBaseModel
fromtypingimportNewType
# 定义业务类型
PhoneNumber = NewType('PhoneNumber', str)
Email = NewType('Email', str)
classContact(BaseModel):
phone: PhoneNumber
email: Email
@validator('phone')
defvalidate_phone(cls, v):
ifnotv.startswith('+86'):
raiseValueError('必须是中国的手机号')
returnv3. 模型导出和导入:
# 导出为字典
user = User(name='张三', age=25)
data = user.dict() # {'name': '张三', 'age': 25}
json_str = user.json() # JSON字符串
# 从字典创建
user2 = User.parse_obj({'name': '李四', 'age': 30})
# 排除默认值
user.dict(exclude_defaults=True)用了pydantic之后,我最大的感受是:写Python代码终于有了“安全感”。不再担心某个接口传错了参数,不再纠结配置文件的管理,不再重复定义相似的数据类。
有时候,选择一个好的工具,比写一万行代码更重要。pydantic就是这样一个工具——它不会改变Python的动态特性,但能让你的代码更加健壮、可维护。
如果你还在手动写参数验证,还在用全局变量管理配置,还在重复定义数据类……那么,是时候试试pydantic了。相信我,用完之后你会和我一样感慨:为什么没早点知道这个库!
“无他,惟手熟尔”!有需要的用起来!
本文分享自 Nicholas与Pypi 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!