将 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

正如 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()

的工作方式几乎相同。

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

Answer 2

您可能想要使用自定义 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 3

您可以使用不同的Python包（例如orjson）来为pydantic模型配置json。例如

import orjson

class User(BaseModel):
    username: str

    class Config:
        orm_mode = True
        json_loads = orjson.loads
        json_dumps = orjson.dumps
        arbitrary_types_allowed = True