SQLAlchemy中json列的自定义JSON序列化程序

Question

我有以下ORM对象(简化)：

import datetime as dt

from sqlalchemy import create_engine, Integer, Column, DateTime
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Session, declarative_base

Base = declarative_base()

class Metrics(Base):
    __tablename__ = 'metrics'

    id = Column(Integer, primary_key=True)
    ts = Column(DateTime, default=dt.datetime.now())
    computed_values = Column(JSONB)
    dates = Column(JSONB)


entry = Metrics(computed_values={'foo': 12.3, 'bar':45.6},
                dates=[dt.date.today()])

engine = create_engine('postgresql://postgres:postgres@localhost:5432/my_schema')
with Session(engine, future=True) as session:
    session.add(entry)
    session.commit()

每一行都有：

• id主键
• 插入行时的ts时间戳
• 要存储的computed_values实际JSONB数据
• dates JSONB存储计算数据的日期列表。

虽然我对computed_values列没有问题，但是datetime.date列中的datetime.date对象在默认情况下不能序列化SQLAlchemy JSON序列化程序。

我的想法是为该列重新定义date对象的序列化程序行为。为此，我必须定义自己的自定义JSON序列化程序，或者使用一些现成的JSON序列化程序，比如orjson。由于我可能会在项目中遇到许多其他JSON序列化问题，所以我更喜欢后者。

深入研究JSONB类和它的超类，我认为下面的技巧应该是正确的：

class Metrics(Base):
    __tablename__ = 'metrics'

    # ---%<--- snip ---%<---
    dates = Column(JSONB(json_serializer=lambda obj: orjson.dumps(obj, option=orjson.OPT_NAIVE_UTC)))

    # ---%<--- snip ---%<---

但事实并非如此：

File "metrics.py", line 30, in Metrics
    dates = Column(JSONB(json_serializer=lambda obj: orjson.dumps(obj, option=orjson.OPT_NAIVE_UTC)))
TypeError: __init__() got an unexpected keyword argument 'json_serializer'

我做错了什么，如何正确定义JSON (和JSONB)列的自定义SQLAlchemy序列化程序？

Answer 1

看起来，您应该能够通过修改create_engine语句来获得所需的东西。

来自SQLAlchemy中的docstring

Custom serializers and deserializers are specified at the dialect level,
that is using :func:`_sa.create_engine`.  The reason for this is that when
using psycopg2, the DBAPI only allows serializers at the per-cursor
or per-connection level.   E.g.::
    engine = create_engine("postgresql://scott:tiger@localhost/test",
                            json_serializer=my_serialize_fn,
                            json_deserializer=my_deserialize_fn
                    )

因此，由此产生的代码应该如下：

import datetime as dt

import orjson

from sqlalchemy import create_engine, Integer, Column, DateTime
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Session, declarative_base

Base = declarative_base()

class Metrics(Base):
    __tablename__ = 'metrics'

    id = Column(Integer, primary_key=True)
    ts = Column(DateTime, default=dt.datetime.now())
    computed_values = Column(JSONB)
    dates = Column(JSONB)

entry = Metrics(computed_values={'foo': 12.3, 'bar':45.6},
                dates=[dt.date.today()])

def orjson_serializer(obj):
    """
        Note that `orjson.dumps()` return byte array, while sqlalchemy expects string, thus `decode()` call.
    """
    return orjson.dumps(obj, option=orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NAIVE_UTC).decode()


engine = create_engine('postgresql://postgres:postgres@localhost:5432/my_schema',
    json_serializer=orjson_serializer,
    json_deserializer=orjson.loads)
with Session(engine, future=True) as session:
    session.add(entry)
    session.commit()