Writing a pydantic object into a sqlalchemy json column

I’m looking for a way to have a pydantic object stored in a sqlalchemy json column. My attempts so far are being tripped up by a datetime field in the pydantic object. I feel like I’m missing something obvious.

My first attempt was to simply serialise the result of .dict(). But this doesn’t convert datetime objects to strings so the serialiser falls over. If I convert with .json then the result is a string and what’s stored in the database is the json of a string not a dict.

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

As Eduard Sukharev describes in his answer, you can set sqlalchemy to use a different json encoder.

It’s really well buried, but pydantic does give you access to it’s own json encoders which handle things like datetime automatically

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)

… Then create a sqlalchemy engine with:

create_engine(conn_string, json_serializer=_custom_json_serializer)

With that sqlalchemy will be able to handle .dict() results in pretty much the same way pydantic .json() works.

Note this doesn’t work for classes with their own custom encoders.