我有以下课程:

class Thing:
    def __init__(self, x: str):
        self.x = x

    def __str__(self):
        return self.x

    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, v: str) -> "Thing":
        return cls(v)

由于验证器方法,我可以将此类用作Pydatics模型中的自定义字段类型:

from pydantic import BaseModel
from thing import Thing

class Model(BaseModel):
    thing: Thing

但是,如果我想要序列化到JSON,我需要在Pydtic模型上设置json_encoders选项:

class Model(BaseModel):
    class Config:
        json_encoders = {
             Thing: str
        }
    thing: Thing

现在,皮丹蒂克可以将Thing序列化为JSON,然后再序列化为JSON.但配置在两个地方:一部分在Model,一部分在Thing级.我想全部押在Thing英镑上.

有没有办法在Thing上设置json_encoders选项,以便皮丹蒂克知道如何透明地处理它?

注意,Thing在这里被最小化了:它有很多逻辑,我并不只是试图声明一个定制的str类型.

推荐答案

在我看来,这实际上是一个比平庸模型更深层次的问题.我找到了this ongoing discussion个关于是否应该在Python中引入带有__json____serialize__方法的标准协议的问题.

问题在于,在定制类型的编码/序列化逻辑与类本身分离的情况下,PYDANIC受到标准库json模块的相同限制.

无论引入这样一个协议的更广泛的 idea 是否有意义,我们都可以稍微借鉴一下它来定义一个定制的json.dumps版本,该版本判断是否存在__serialize__方法,并将其用作default函数来序列化对象.(有关default参数的说明,请参阅json.dump文档.)

然后,我们可以设置一个自定义基本型号,并将Config.json_dumps选项设置为该功能.这样,所有子模型将自动回退到用于序列化的子模型(例如,除非被BaseModel.json方法的encoder参数覆盖).

下面是一个例子:

base.py

from collections.abc import Callable
from json import dumps as json_dumps
from typing import Any

from pydantic import BaseModel as PydanticBaseModel


def json_dumps_extended(obj: object, **kwargs: Any) -> str:
    default: Callable[[object], object] = kwargs.pop("default", lambda x: x)

    def custom_default(to_encode: object) -> object:
        serialize_method = getattr(to_encode, "__serialize__", None)
        if serialize_method is None:
            return default(to_encode)
        return serialize_method()  # <-- already bound to `to_encode`

    return json_dumps(obj, default=custom_default, **kwargs)


class BaseModel(PydanticBaseModel):
    class Config:
        json_dumps = json_dumps_extended

application.py

from __future__ import annotations
from collections.abc import Callable, Iterator

from .base import BaseModel


class Thing:
    def __init__(self, x: str) -> None:
        self.x = x

    def __str__(self) -> str:
        return self.x

    def __serialize__(self) -> str:  # <-- this is the magic method
        return self.x

    @classmethod
    def __get_validators__(cls) -> Iterator[Callable[..., Thing]]:
        yield cls.validate

    @classmethod
    def validate(cls, v: str) -> Thing:
        return cls(v)


class Model(BaseModel):
    thing: Thing
    num: float = 3.14


instance = Model(thing=Thing("foo"))
print(instance.json(indent=4))

输出:

{
    "thing": "foo",
    "num": 3.14
}

给Python<3.9用户的注意事项:从typing而不是collections.abc导入CallableIterator类型.


PS

如果您希望能够在除基本模型之外的更多地方重新使用这种方法进行序列化,那么在类型上多花点功夫可能是个好主意.我们的__serialize__方法的runtime_checkable个定制协议可能会很有用.

此外,我们还可以通过使用functools.partial使json_dumps_extended方法不那么笨重.

Here is a slightly more sophisticated version of the suggested base.py个:

from collections.abc import Callable
from functools import partial
from json import dumps as json_dumps
from typing import Any, Optional, Protocol, TypeVar, overload, runtime_checkable

from pydantic import BaseModel as PydanticBaseModel

T = TypeVar("T")
T_co = TypeVar("T_co", covariant=True)
Func1Arg = Callable[[object], T]


@runtime_checkable
class Serializable(Protocol[T_co]):
    def __serialize__(self) -> T_co: ...


@overload
def serialize(obj: Serializable[T_co]) -> T_co: ...


@overload
def serialize(obj: Any, fallback: Func1Arg[T]) -> T: ...


def serialize(obj: Any, fallback: Optional[Func1Arg[Any]] = None) -> Any:
    if isinstance(obj, Serializable):
        return obj.__serialize__()
    if fallback is None:
        raise TypeError(f"Object not serializable: {obj}")
    return fallback(obj)


def _id(x: T) -> T: return x


def json_dumps_extended(obj: object, **kwargs: Any) -> str:
    custom_default = partial(serialize, fallback=kwargs.pop("default", _id))
    return json_dumps(obj, default=custom_default, **kwargs)


class BaseModel(PydanticBaseModel):
    class Config:
        json_dumps = json_dumps_extended

另一种 Select 可能是直接修补JSONEncoder.default个猴子.但是,在没有进一步配置的情况下,Pydtic似乎仍然自己执行类型判断,甚至在调用该方法之前就阻止了序列化.

我不认为我们有更好的 Select ,除非引入一些标准的序列化协议(至少对于JSON).

Python相关问答推荐

Pandas或pyspark跨越列创建

有没有方法可以关闭Python多处理资源跟踪器进程?

aiohTTP与pytest的奇怪行为

在Python中添加期货之间的延迟

如何使用函数正确索引收件箱?

如何才能将每个组比上一组增加N %?

如何使用Python中的clinicalTrials.gov API获取完整结果?

如何计算两极打印机中 * 所有列 * 的出现次数?

根据不同列的值在收件箱中移动数据

如何访问所有文件,例如环境变量

从numpy数组和参数创建收件箱

如何找到满足各组口罩条件的第一行?

如何让程序打印新段落上的每一行?

切片包括面具的第一个实例在内的眼镜的最佳方法是什么?

根据二元组列表在pandas中创建新列

修复mypy错误-赋值中的类型不兼容(表达式具有类型xxx,变量具有类型yyy)

如何在solve()之后获得症状上的等式的值

在pandas中使用group_by,但有条件

在单个对象中解析多个Python数据帧

解决调用嵌入式函数的XSLT中表达式的语法移位/归约冲突