如何用Pydantic验证自定义根类型，完成自定义类/字面量的可辨识联合校验

现有实现缺陷 #

1. 开启arbitrary_types_allowed后Pydantic只会跳过未知类型的校验，不会验证传入的类是否属于R1/R2，无法实现校验效果。
2. 直接将根模型作为函数参数类型，调用时传入原生类/字符串不会触发自动校验，会报类型不匹配错误。
3. 动作类型的字面量定义和注释里的预期允许值right不一致，会导致合法请求被拦截。

最优解决方案（适配Pydantic v2） #

用自定义校验类型+validate_call装饰器的组合，无需修改调用习惯，自动完成参数校验：

完整实现代码 #

# authorization.py
from typing import Any, Union, Literal, Type
from pydantic import validate_call, GetCoreSchemaHandler
from pydantic_core import core_schema
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base

Base: DeclarativeMeta = declarative_base()

class R1(Base):
    __tablename__ = "r1"
    # 你的表字段定义


class R2(Base):
    __tablename__ = "r2"
    # 你的表字段定义


# 自定义资源类型校验器，校验输入是否为R1/R2类对象
class ResourceType:
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        def validate_resource(v: Any) -> Type[Union[R1, R2]]:
            if v is R1 or v is R2:
                return v
            raise ValueError(f"仅支持R1、R2类作为资源参数，当前输入为{repr(v)}")
        
        return core_schema.no_info_after_validator_function(
            validate_resource,
            core_schema.any_schema()
        )

# 动作类型直接用字面量定义，需要支持right就修改为 Literal["read", "right"]
ActionType = Literal["read", "edit"]


@validate_call
async def get_authorized_resources(resource: ResourceType):
    # 此处resource已校验为R1/R2类对象，可直接使用
    pass


@validate_call
async def is_allowed(resource: ResourceType, action: ActionType):
    # 此处resource和action均已完成校验
    pass

调用方式（完全符合预期） #

# another_file.py
from authorization import R1, R2, is_allowed

# 无需额外封装，直接传值即可
await is_allowed(resource=R1, action="read")

兼容方案（保留原有根模型定义） #

如果必须保留R和A两个根模型的写法，调整如下：

class R(BaseModel):
    __root__: ResourceType

    class Config:
        arbitrary_types_allowed = True


class A(BaseModel):
    __root__: Literal["read", "edit"]


# 调用时手动实例化根模型即可
await is_allowed(resource=R(__root__=R1), action=A(__root__="read"))

Pydantic v1适配说明 #

如果使用Pydantic v1版本，自定义类型的校验逻辑替换为__get_validators__方法即可：

class ResourceType:
    @classmethod
    def __get_validators__(cls):
        yield cls.validate_resource

    @classmethod
    def validate_resource(cls, v: Any) -> Type[Union[R1, R2]]:
        if v is R1 or v is R2:
            return v
        raise ValueError(f"仅支持R1、R2类作为资源参数，当前输入为{repr(v)}")

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