如何正确创建Pydantic扩展模型？FastAPI+ES遇验证失败

我正在开发一个基于FastAPI和ElasticSearch的应用，希望通过抽象化的Pydantic BaseModel创建实体类，实现API层与索引层的互操作性。但在继承BaseIndex（继承自BaseModel的抽象基类）创建UserIndex子类时，实例化对象时触发验证错误，提示created_on、modified_on、created_by、modified_by字段缺失。

类层级代码 #

from abc import ABCMeta, abstractmethod
from datetime import datetime
from pydantic import BaseModel, field_serializer, field_validator

class BaseIndex(BaseModel, metaclass=ABCMeta):
    id: str
    name: str
    created_on: datetime
    modified_on: datetime
    created_by: str
    modified_by: str

    @field_serializer("created_on", "modified_on", mode="plain")
    @classmethod
    def serialize_datetime(cls, v: datetime):
        return int(v.timestamp())

    @field_validator("created_on", "modified_on", mode="after")
    @classmethod
    def validate_datetime(cls, v: int):
        return datetime.utcfromtimestamp(v)

    @abstractmethod
    def as_schema(self, with_defaults: bool = False):
        pass

    @classmethod
    def from_record(cls, record: dict[str, Any]):
        pass
# ----------------------------------------------------------- #
class UserIndex(BaseIndex):
    user_name: str
    email: str
    first_name: str
    last_name: str
    groups: list[str]
    roles: list[str]
    last_login: datetime

    @field_serializer("last_login", mode="plain")
    def serialize_last_login(cls, value: datetime):
        return int(value.timestamp())

    @field_validator("last_login", mode="after")
    def validate_last_login(cls, value: int):
        return datetime.utcfromtimestamp(value)

    def as_schema(self, with_defaults: bool = False):
        return {
            "properties": {
                "id": {"type": "keyword"},
                "name": {"type": "string"},
                "user_name": {"type": "keyword"},
                "email": {"type": "keyword"},
                "first_name": {"type": "string", "copy_to": "name"},
                "last_name": {"type": "string", "copy_to": "name"},
                "groups": {"type": "string"},
                "roles": {"type": "string"}
            },

        }

实例化代码 #

user = UserIndex(id="u1", name="A1", user_name="Bob", email="bob@bb.com", first_name="Bob", last_name="B", 
            groups=["g1", "g2"], roles=["r1", "r2"], last_login=datetime.now())

触发的验证错误 #

File ".....\Lib\site-packages\pydantic\main.py", line 150, in __init__
    __pydantic_self__.__pydantic_validator__.validate_python(data, self_instance=__pydantic_self__)
pydantic_core._pydantic_core.ValidationError: 4 validation errors for User
created_on
  Field required [type=missing, input_value={'id': 'u1..': 'A1'}, input_type=dict]
modified_on
  Field required [type=missing, input_value={'id': 'u1..': 'A1'}, input_type=dict]
created_by
  Field required [type=missing, input_value={'id': 'u1..': 'A1'}, input_type=dict]
modified_by
  Field required [type=missing, input_value={'id': 'u1..': 'A1'}, input_type=dict]

1. 为BaseIndex的必填字段设置默认值 #

BaseIndex中的四个字段是必填定义，实例化子类时必须传入。可以通过Field的default_factory或直接设置默认值，实现自动填充：

from abc import ABCMeta, abstractmethod
from datetime import datetime
from pydantic import BaseModel, field_serializer, field_validator, Field
from typing import Any

class BaseIndex(BaseModel, metaclass=ABCMeta):
    id: str
    name: str
    # 用default_factory动态生成UTC时间，避免所有实例共享同一时间戳
    created_on: datetime = Field(default_factory=datetime.utcnow)
    modified_on: datetime = Field(default_factory=datetime.utcnow)
    # 根据业务场景设置默认值，示例用"system"
    created_by: str = Field(default="system")
    modified_by: str = Field(default="system")

    @field_serializer("created_on", "modified_on", mode="plain")
    @classmethod
    def serialize_datetime(cls, v: datetime):
        return int(v.timestamp())

    @field_validator("created_on", "modified_on", mode="before")
    @classmethod
    def validate_datetime(cls, v):
        # 兼容输入为时间戳的场景，先转换为datetime再验证
        if isinstance(v, int):
            return datetime.utcfromtimestamp(v)
        return v

    @abstractmethod
    def as_schema(self, with_defaults: bool = False):
        pass

    @classmethod
    def from_record(cls, record: dict[str, Any]):
        # 实现从ES记录转换为模型的逻辑，自动处理时间戳字段
        record_copy = record.copy()
        time_fields = ["created_on", "modified_on", "last_login"]
        for field in time_fields:
            if field in record_copy and isinstance(record_copy[field], int):
                record_copy[field] = datetime.utcfromtimestamp(record_copy[field])
        return cls(**record_copy)

2. 修正子类的序列化/验证方法 #

UserIndex中的序列化方法应为实例方法（参数用self），验证方法需添加@classmethod装饰器，并将mode改为before（确保在字段验证前完成类型转换）：

class UserIndex(BaseIndex):
    user_name: str
    email: str
    first_name: str
    last_name: str
    groups: list[str]
    roles: list[str]
    last_login: datetime = Field(default_factory=datetime.utcnow)

    @field_serializer("last_login", mode="plain")
    def serialize_last_login(self, value: datetime):
        return int(value.timestamp())

    @field_validator("last_login", mode="before")
    @classmethod
    def validate_last_login(cls, value):
        if isinstance(value, int):
            return datetime.utcfromtimestamp(value)
        return value

    def as_schema(self, with_defaults: bool = False):
        return {
            "properties": {
                "id": {"type": "keyword"},
                "name": {"type": "string"},
                "user_name": {"type": "keyword"},
                "email": {"type": "keyword"},
                "first_name": {"type": "string", "copy_to": "name"},
                "last_name": {"type": "string", "copy_to": "name"},
                "groups": {"type": "string"},
                "roles": {"type": "string"},
                "created_on": {"type": "date", "format": "epoch_second"},
                "modified_on": {"type": "date", "format": "epoch_second"},
                "last_login": {"type": "date", "format": "epoch_second"}
            },
        }

3. 实例化子类 #

修改后，实例化时无需手动传入四个基础字段，可直接使用默认值，也可按需覆盖：

from datetime import datetime

user = UserIndex(
    id="u1",
    name="A1",
    user_name="Bob",
    email="bob@bb.com",
    first_name="Bob",
    last_name="B",
    groups=["g1", "g2"],
    roles=["r1", "r2"],
    # 可选：手动指定last_login，否则用默认的当前UTC时间
    last_login=datetime.now()
)

补充：动态填充用户字段（FastAPI场景） #

如果created_by/modified_by需要使用当前登录用户，可结合FastAPI的依赖注入实现：

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 实现获取当前登录用户的逻辑
    return {"username": "current_user"}

@app.post("/users/")
async def create_user(
    user_data: UserIndex,
    current_user = Depends(get_current_user)
):
    # 覆盖默认的创建/修改人字段
    user_data.created_by = current_user["username"]
    user_data.modified_by = current_user["username"]
    # 后续逻辑：将数据同步到ElasticSearch
    return user_data

内容的提问来源于stack exchange，提问作者Kris