如何在Pydantic父类静态方法中访问子类属性实现通用CRUD？

解决方案 #

核心思路是把静态方法替换为类方法，类方法的第一个参数cls会自动指向当前调用的子类，可直接访问子类的Meta类属性，完美解决静态方法无法获取子类信息的问题。

完整基类实现 #

from typing import List, Optional, Type, TypeVar
from uuid import uuid4
from pydantic import BaseModel, Field
from server.database import get_db

# 定义泛型绑定，保证类型提示正确
T = TypeVar('T', bound='BaseDBModel')

class BaseDBModel(BaseModel):
    # 公共字段可以直接定义在基类，不用每个子类重复写
    id: Optional[str] = Field(default_factory=lambda: str(uuid4()))

    def __init_subclass__(cls, **kwargs):
        """子类初始化钩子，强制要求配置集合名，避免漏写报错"""
        super().__init_subclass__(**kwargs)
        if not hasattr(cls, 'Meta') or not hasattr(cls.Meta, 'collection_name'):
            raise NotImplementedError(f"{cls.__name__}必须定义Meta类，且配置collection_name属性")

    @classmethod
    async def all(cls: Type[T]) -> List[T]:
        ret = get_db()[cls.Meta.collection_name].find()
        return [cls(**i) async for i in ret]

    @classmethod
    async def get(cls: Type[T], id: str) -> Optional[T]:
        ret = await get_db()[cls.Meta.collection_name].find_one({'id': id})
        return cls(**ret) if ret else None

    async def save(self):
        await get_db()[self.__class__.Meta.collection_name].insert_one(self.dict())

    async def update(self, **kwargs):
        await get_db()[self.__class__.Meta.collection_name].update_one(
            {'id': self.id},
            {'$set': kwargs},
        )

    async def delete(self):
        """还可以按需扩展其他通用CRUD方法"""
        await get_db()[self.__class__.Meta.collection_name].delete_one({'id': self.id})

子类使用示例 #

所有业务模型仅需定义业务字段和集合名即可，无需重复编写CRUD逻辑：

class Campaign(BaseDBModel):
    name: str
    class Meta:
        collection_name = 'campaigns'

class Record(BaseDBModel):
    data: str
    user_id: str
    class Meta:
        collection_name = 'records'

class Fragment(BaseDBModel):
    content: str
    campaign_id: str
    class Meta:
        collection_name = 'fragments'

调用示例 #

# 查询所有Campaign
all_campaigns = await Campaign.all()
# 查询单个Record
record = await Record.get("xxxx-xxxx-xxxx")
# 保存新Fragment
new_frag = Fragment(content="test", campaign_id="xxx")
await new_frag.save()
# 更新Campaign
campaign = await Campaign.get("xxx")
await campaign.update(name="新名称")

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