Python使用Pydantic验证和解析配置数据的完整指南

config = {
    "host": "localhost",
    "port": "5432",  # 错误：应为整数
    "username": "admin",
    "password": "123",  # 错误：密码长度不足
    "timeout": -10     # 错误：超时时间不能为负
}

if "host" not in config:
    raise ValueError("Missing host")
if not isinstance(config["host"], str):
    raise TypeError("Host must be string")
    
if "port" not in config:
    raise ValueError("Missing port")
try:
    port = int(config["port"])
except ValueError:
    raise TypeError("Port must be integer")
if port <= 0 or port > 65535:
    raise ValueError("Port out of range")
    
# 类似验证需要重复编写20+行代码...

try:
    # 执行上述验证
except Exception as e:
    print(f"Config error: {str(e)}")  # 输出: "Config error: Port must be integer"

from pydantic import BaseModel, Field, ValidationError
from typing import Optional

class DatabaseConfig(BaseModel):
    host: str = Field(..., description="数据库主机地址")
    port: int = Field(..., gt=0, le=65535, description="端口号")
    username: str
    password: str = Field(..., min_length=8, description="密码至少8位")
    timeout: float = Field(5.0, gt=0, description="连接超时时间(秒)")
    pool_size: Optional[int] = Field(None, ge=1, description="连接池大小")

try:
    config = DatabaseConfig(
        host="localhost",
        port="5432",  # 字符串自动转为整数
        username="admin",
        password="123",  # 会触发验证错误
        timeout="-10"   # 会触发验证错误
    )
except ValidationError as e:
    print(e.json(indent=2))

[  {    "loc": ["password"],
    "msg": "ensure this value has at least 8 characters",
    "type": "value_error.min_length"
  },
  {
    "loc": ["timeout"],
    "msg": "ensure this value is greater than 0",
    "type": "greater_than"
  }
]

from typing import List

class DataSourceConfig(BaseModel):
    name: str
    table: str
    primary_key: str

class AppConfig(BaseModel):
    database: DatabaseConfig
    data_sources: List[DataSourceConfig]
    debug_mode: bool = False

config_data = {
    "database": {
        "host": "prod-db",
        "port": "5432",
        "username": "app_user",
        "password": "securepass123",
        "timeout": 3.0
    },
    "data_sources": [
        {"name": "users", "table": "sys_users", "primary_key": "id"},
        {"name": "orders", "table": "sys_orders", "primary_key": "order_id"}
    ],
    "debug_mode": "true"  # 会触发类型错误
}

try:
    app_config = AppConfig(**config_data)
except ValidationError as e:
    print(e.json(indent=2))

from pydantic import field_validator

class EnhancedDatabaseConfig(DatabaseConfig):
    @field_validator("host")
    @classmethod
    def validate_host(cls, v):
        if v.startswith("http://") or v.startswith("https://"):
            raise ValueError("Database host should not contain protocol")
        return v.lower()  # 自动转为小写

    @field_validator("password")
    @classmethod
    def validate_password_strength(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain at least one uppercase letter")
        return v

# .env文件内容：
# DB_HOST=localhost
# DB_PORT=5432
# DB_USERNAME=admin
# DB_PASSWORD=P@ssw0rd
# DB_TIMEOUT=3.5

from pydantic_settings import BaseSettings

class EnvConfig(BaseSettings):
    db_host: str
    db_port: int
    db_username: str
    db_password: str
    db_timeout: float = 5.0

    class Config:
        env_file = ".env"  # 自动加载.env文件
        env_prefix = "db_"  # 环境变量前缀

config = EnvConfig()
print(config.db_host)  # 输出: localhost

from pydantic import create_json_schema

schema = create_json_schema(DatabaseConfig)
print(schema)

{
  "title": "DatabaseConfig",
  "type": "object",
  "properties": {
    "host": {"title": "Host", "type": "string"},
    "port": {
      "title": "Port",
      "type": "integer",
      "minimum": 1,
      "maximum": 65535
    },
    "password": {
      "title": "Password",
      "type": "string",
      "minLength": 8
    }
  },
  "required": ["host", "port", "username", "password"]
}