将 pydantic 对象写入 sqlalchemy json 列

Question

我正在寻找一种将 pydantic 对象存储在 sqlalchemy json 列中的方法。到目前为止，我的尝试被 pydantic 对象中的 datetime 字段绊倒了。我觉得我错过了一些明显的东西。

我的第一次尝试是简单地序列化 .dict() 的结果。但这不会将日期时间对象转换为字符串，因此序列化程序会失败。如果我用 .json 转换，那么结果是一个字符串，数据库中存储的是字符串的 json 而不是字典。

import sqlalchemy.orm
from pydantic import BaseModel
from datetime import datetime

mapper_registry = sqlalchemy.orm.registry()
Base = mapper_registry.generate_base()


class _PydanticType(sqlalchemy.types.TypeDecorator):
    impl = sqlalchemy.types.JSON

    def __init__(self, pydantic_type):
        super().__init__()
        self._pydantic_type = pydantic_type

    def process_bind_param(self, value, dialect):
        return value.dict() if value else None

    def process_result_value(self, value, dialect):
        return self._pydantic_type.parse_obj(value) if value else None


class Test(BaseModel):
    timestamp: datetime


class Foo(Base):
    __tablename__ = 'foo'
    x = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
    y = sqlalchemy.Column(_PydanticType(Test))


engine = sqlalchemy.create_engine('sqlite:///x.db', echo=True)
mapper_registry.metadata.create_all(bind=engine)
session = sqlalchemy.orm.sessionmaker(bind=engine)()
session.add(Foo(x=1, y=Test(timestamp=datetime.now())))
session.commit()

sqlalchemy.exc.StatementError: (builtins.TypeError) Object of type datetime is not JSON serializable

Answer 1

您可能想要使用自定义 json 序列化程序，例如 orjson，它可以为您优雅地处理日期时间 [反] 序列化。

只需将序列化回调作为 json_serializer 参数传递给 create_engine():

# orjson.dumps returns bytearray, so you'll can't pass it directly as json_serializer
def _orjson_serializer(obj):
    # mind the .decode() call
    # you can also define some more useful options
    return orjson.dumps(obj, option=orjson.OPT_NAIVE_UTC).decode()

create_engine(conn_string, json_serializer=_orjson_serializer)

有关传递给 orjson.dumps 的更多选项，请参阅 orjson README。

Answer 2

如 Eduard Sukharev 在中所述，您可以将 sqlalchemy 设置为使用不同的 json 编码器。

它隐藏得很好，但 pydantic 确实让您可以访问它自己的 json 编码器，它可以自动处理诸如日期时间之类的事情

import json
import pydantic.json


def _custom_json_serializer(*args, **kwargs) -> str:
    """
    Encodes json in the same way that pydantic does.
    """
    return json.dumps(*args, default=pydantic.json.pydantic_encoder, **kwargs)

... 然后创建一个 sqlalchemy 引擎：

create_engine(conn_string, json_serializer=_custom_json_serializer)

有了这个，sqlalchemy 将能够以几乎与 pydantic .json() 工作方式相同的方式处理 .dict() 结果。

请注意，这不适用于类使用他们自己的自定义编码器。

将 pydantic 对象写入 sqlalchemy json 列

Writing a pydantic object into a sqlalchemy json column

python

datetime

sqlalchemy

pydantic