简介

python类型注释的数据校验和设置管理 pydantic在运行时强制执行类型提示，并在数据无效时提供友好的错误 换句话说，pydantic保证输出模型的类型和约束，而不是输入数据 在pydantic中定义对象的主要方法是通过继承自BaseModel类的模型 官方文档：https://pydantic-docs.helpmanual.io/

QuickStart

  1. 安装

    1. pip install pydantic

  2. case ```python from datetime import datetime from typing import List, Optional from pydantic import BaseModel, ValidationError

class User(BaseModel): id: int name = “zaygee” signup_ts: Optional[datetime] = None friends: List[int] = []

external_data = { ‘id’: ‘123’, ‘signup_ts’: ‘2019-06-01 12:22’, ‘friends’: [1, 2, ‘3’], }

字典解包

user = User(**external_data) print(user.id) # 123

try: User(signup_ts = “broken”, friends=[1, 2, “not number”]) except ValidationError as e: print(e.json())

[

{

“loc”: [

“id”

],

“msg”: “field required”,

“type”: “value_error.missing”

},

{

“loc”: [

“signup_ts”

],

“msg”: “invalid datetime format”,

“type”: “value_error.datetime”

},

{

“loc”: [

“friends”,

2

],

“msg”: “value is not a valid integer”,

“type”: “type_error.integer”

}

]

  1. <a name="LydxY"></a>
  2. ### 模型&模型属性
  3. 基本模型继承自类 BaseModel
  4. <a name="GsCif"></a>
  5. #### dict(): 返回模型字段和值的字典
  6. > - include: 包含在返回字典中的字段
  7. > -  exclude：要从返回的字典中排除的字段
  8. > - by_alias: 是否应使用字段别名作为返回字典中的键
  9. > - exclude_unset: 创建模型时未明确设置的字段是否应从返回的字典中排除, 默认 False
  10. > -  exclude_defaults：是否应从返回的字典中排除等于其默认值（无论是否设置）的字段,默认 False
  11. > - exclude_none: 是否None应该从返回的字典中排除等于的字段；默认 False
  13. ```python
  14. print("返回模型字段和值的字典", user.dict(include={"name"}, exclude_unset=True))    # {'name': 'zaygee'}
  15. print(user.dict(exclude_defaults=True))
  16. print(user.dict(exclude_defaults=False))

json(): 会将模型序列化为 JSON

  • include: 包含在返回字典中的字段
  • exclude：要从返回的字典中排除的字段
  • by_alias: 是否应使用字段别名作为返回字典中的键
  • exclude_unset: 创建模型时未明确设置的字段是否应从返回的字典中排除, 默认 False
  • exclude_defaults：是否应从返回的字典中排除等于其默认值（无论是否设置）的字段,默认 False
  • exclude_none: 是否None应该从返回的字典中排除等于的字段；默认 False
  • encoder：传递给default参数的自定义编码器函数json.dumps()；默认为自定义编码器，旨在处理所有常见类型
  • dumps_kwargs: 任何其他关键字参数都传递给json.dumps()，例如indent
  1. print("返回json字符串表示dict():", user.json(indent=4), type(user.json()))

copy(): 浅复制模型副本

  • 典中的字段
  • exclude：要从返回的字典中排除的字段
  • update：创建复制模型时要更改的值字典
  • deep: 是否对新模型进行深拷贝；默认False
  1. print("浅复制模型副本:", user.copy(include={"name", "id"}))  # id=123 name='zaygee1'
  2. print(user.copy(update={"name": "update_name"}))    # id=123 signup_ts=datetime.datetime(2019, 6, 1, 12, 22) friends=[1, 2, 3] name='update_name'
  3. print("排除字段exclude:", user.copy(exclude={"name", "signup_ts"}))  # id=123 friends=[1, 2, 3]
  4. print("deep深复制:", user.copy(deep=True))

parse_obj(): 如果对象不是字典，则用于将任何对象加载到模型中并进行错误处理的实用程序

  1. print(user.parse_obj({"id": 123,"signup_ts":1234567890,"name":"John Doe"}) )    # id=123 signup_ts=datetime.datetime(2009, 2, 13, 23, 31, 30, tzinfo=datetime.timezone.utc) friends=[] name='John Doe'
  2. # print(user.parse_obj('{"id": 123,"signup_ts":1234567890,"name":"John Doe"}') )  # validation error for User

parse_raw(): 用于加载多种格式的字符串的实用程序

  1. print(user.parse_raw('{"id": 123,"signup_ts":1234567890,"name":"John Doe"}'))

parse_file(): 就像parse_raw()但对于文件路径

schema(): 返回将模型表示为 JSON 模式的字典

  1. print("-"*50)
  2. print(user.schema())
  3. print("-"*50)
  5. # {'title': 'User', 'type': 'object', 'properties': {'id': {'title': 'Id', 'type': 'integer'}, 'signup_ts': {'title': 'Signup Ts', 'type': 'string', 'format': 'date-time'}, 'friends': {'title': 'Friends', 'default': [], 'type': 'array', 'items': {'type': 'integer'}}, 'name': {'title': 'Name', 'default': 'zaygee', 'type': 'string'}}, 'required': ['id']}

schema_json(): 返回 JSON 字符串表示schema()

  1. print("-"*50)
  2. print(user.schema_json())
  3. print("-"*50)
  5. # --------------------------------------------------
  6. # {"title": "User", "type": "object", "properties": {"id": {"title": "Id", "type": "integer"}, "signup_ts": {"title": "Signup Ts", "type": "string", "format": "date-time"}, "friends": {"title": "Friends", "default": [], "type": "array", "items": {"type": "integer"}}, "name": {"title": "Name", "default": "zaygee", "type": "string"}}, "required": ["id"]}
  7. # --------------------------------------------------

fields_set: 初始化模型实例时设置的字段名称集

  1. print(user.__fields_set__)
  2. # {'name', 'signup_ts', 'friends', 'id'}

fields: 模型字段的字典

  1. print(user.__fields__)
  2. # {'id': ModelField(name='id', type=int, required=True), 'signup_ts': ModelField(name='signup_ts', type=Optional[datetime], required=False, default=None), 'friends': ModelField(name='friends', type=List[int], required=False, default=[]), 'name': ModelField(name='name', type=str, required=False, default='zaygee')}

错误处理

字段类型校验

必填字段Field(…)&可选字段Optional[]

  1. from pydantic import Field
  2. from typing import Optional
  4. # a、b、c都是必选的，但是c可填None
  5. class Model(BaseModel):
  6.     a: int
  7.     c: int = Field(...)
  8.     d: Optional[int] = Field(...)
  10. print(Model(a=0, c=2, d=None))

parse_obj_as将字段解析为指定类型

  1. from typing import List
  2. from pydantic import parse_obj_as
  4. class Item(BaseModel):
  5.     id: int
  6.     name: str
  7. item_data = [{'id': 1, 'name': 'My Item'}]
  8. items = parse_obj_as(List[Item], item_data)
  9. print(items, type(items))
  10. for index, key in items:
  11.     print(index, key)
  12.     print(index[-1], key[-1])
  15. # [Item(id=1, name='My Item')] <class 'list'>
  16. # ('id', 1) ('name', 'My Item')
  17. # 1 My Item

字段类型约束

  1. from decimal import Decimal
  3. from pydantic import (
  4.     BaseModel,
  5.     NegativeFloat,
  6.     NegativeInt,
  7.     PositiveFloat,
  8.     PositiveInt,
  9.     NonNegativeFloat,
  10.     NonNegativeInt,
  11.     NonPositiveFloat,
  12.     NonPositiveInt,
  13.     conbytes,
  14.     condecimal,
  15.     confloat,
  16.     conint,
  17.     conlist,
  18.     conset,
  19.     constr,
  20.     Field,
  21.     ByteSize,
  22.     HttpUrl,
  23.     SecretStr,
  24.     FilePath,
  25.     Json
  26. )
  27. class AllModel(BaseModel):
  28.     file_path: FilePath # 文件校验
  29.     url: Optional[HttpUrl]  # 网站校验
  30.     password: SecretStr # 秘文
  31.     lower_str: constr(to_lower=True, min_length=1, max_length=10)   # con*的约束类型
  32.     size: ByteSize  # 字节大小
  34. class ComplexJsonModel(BaseModel):
  35.     json_obj: Json[List[int]]
  37. print(ComplexJsonModel(json_obj='[1, 2, 3]'))
  39. print(AllModel(
  40.     file_path="Script/python_test/base/test_pydantic.py",
  41.     url="https://www.baidu.com",
  42.     password="testpw",
  43.     lower_str="SSS",
  44.     size=520
  45.     ))
  47. #file_path=PosixPath('Script/python_test/base/test_pydantic.py') url=HttpUrl('https://www.baidu.com', scheme='https', host='www.baidu.com', tld='com', host_type='domain') password=SecretStr('**********') lower_str='sss' size=520

验证器-todo

pre & per-item验证器

验证器 & each_item

始终验证

重用验证器

数据类验证器

配置文件加载

  1. from typing import Set
  2. from pydantic import (
  3.     BaseModel,
  4.     BaseSettings,
  5.     PyObject,
  6.     RedisDsn,
  7.     PostgresDsn,
  8.     Field,
  9. )
  11. class SubModel(BaseModel):
  12.     foo = 'bar'
  13.     apple = 1
  15. class Settings(BaseSettings):
  16.     auth_key: str
  17.     api_key: str = Field(..., env='my_api_key')
  19.     redis_dsn: RedisDsn = 'redis://user:pass@localhost:6379/1'
  20.     pg_dsn: PostgresDsn = 'postgres://user:pass@localhost:5432/foobar'
  22.     special_function: PyObject = 'math.cos'
  24.     # to override domains:
  25.     # export my_prefix_domains='["foo.com", "bar.com"]'
  26.     domains: Set[str] = set()
  28.     # to override more_settings:
  29.     # export my_prefix_more_settings='{"foo": "x", "apple": 1}'
  30.     more_settings: SubModel = SubModel()
  32.     class Config:
  33.         env_prefix = 'my_prefix_'  # defaults to no prefix, i.e. ""
  34.         fields = {
  35.             'auth_key': {
  36.                 'env': 'my_auth_key',
  37.             },
  38.             'redis_dsn': {
  39.                 'env': ['service_redis_dsn', 'redis_url']
  40.             }
  41.         }
  43. print(Settings(auth_key="xxx", api_key="xxx").dict())
  45. #{'auth_key': 'xxx', 'api_key': 'xxx', 'redis_dsn': RedisDsn('redis://user:pass@localhost:6379/1', scheme='redis', user='user', password='pass', host='localhost', host_type='int_domain', port='6379', path='/1'), 'pg_dsn': PostgresDsn('postgres://user:pass@localhost:5432/foobar', scheme='postgres', user='user', password='pass', host='localhost', host_type='int_domain', port='5432', path='/foobar'), 'special_function': <built-in function cos>, 'domains': set(), 'more_settings': {'foo': 'bar', 'apple': 1}}

结合dotenv加载配置文件

安装dotenv

  1. # 安装dotenv
  2. pip install python-dotenv
  3. # 新建配置文件
  4. touch .env
  6. ---------------------
  7. # .env 文件
  9. # ignore comment
  10. ENVIRONMENT="production"
  11. REDIS_ADDRESS=localhost:6379
  12. APP_NAME='Hello world'
  13. ---------------------
  15. # 结合Dotenv文件设置变量
  16. from pydantic import BaseSettings
  19. class Settings(BaseSettings):
  20.     app_name: str = "Awesome API"
  21.     environment: str
  22.     items_per_user: int = 50
  24.     class Config:
  25.         env_file = '.env'
  26.         env_file_encoding = "utf-8"
  28. settings = Settings(_env_file='/Users/zaygee/work_script/.env', _env_file_encoding='utf-8')
  29. print(settings) # app_name='Hello world' environment='production' redis_address='localhost:6379'