将 pydantic 对象写入 sqlalchemy json 列

Question

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

我的第一次尝试是简单地序列化.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

正如 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 将能够以与.dict()pydantic 几乎相同的方式处理结果.json()。

请注意，这不适用于具有自己的自定义编码器的类。