为SQLAlchemy的PostgreSQL JSONB实现使用自定义JSON编码器

Question

我正在使用SQLAlchemy的核心库来访问一些PostgreSQL数据库.考虑我有下表:

create table foo (j jsonb);

以下python代码:

from decimal import *
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey, DateTime
from sqlalchemy.dialects.postgresql import JSONB
metadata = MetaData(schema="public")
foo = Table('foo', metadata,Column('f', JSONB))
d = Decimal(2)
ins = foo.insert().values(j = {'d': d})
# assuming engine is a valid sqlalchemy's connection
engine.execute(ins)

最后一句失败,出现以下错误:

StatementError("(builtins.TypeError) Decimal('2') is not JSON serializable",)

这就是我提出这个问题的原因:有没有办法为SQLAchemy指定一个自定义编码器,以便在将json数据编码成PostgreSQL方言时使用？

Answer 1

这通过json_serializer关键字参数支持create_engine,如下所述sqlalchemy.dialects.postgresql.JSON:

def _default(val):
    if isinstance(val, Decimal):
        return str(val)
    raise TypeError()

def dumps(d):
    return json.dumps(d, default=_default)

engine = create_engine(..., json_serializer=dumps)

Answer 2

如果你和我一样，正在寻找一种使用 Flask-SQLAlchemy 运行它的好方法，这就是我所做的。如果您导入并传递flask.json而不是标准库json模块，您将获得日期、日期时间和uuid.UUID实例的自动反序列化。

class HackSQLAlchemy(SQLAlchemy):
    """ Ugly way to get SQLAlchemy engine to pass the Flask JSON serializer
    to `create_engine`.

    See https://github.com/mitsuhiko/flask-sqlalchemy/pull/67/files

    """

    def apply_driver_hacks(self, app, info, options):
        options.update(json_serializer=json.dumps)
        super(HackSQLAlchemy, self).apply_driver_hacks(app, info, options)

Answer 3

如果您使用 Flask，则已经定义了一个扩展的 JSONEncoder，其中flask.json处理UUID，但不处理Decimal。json_serializer它可以使用 @univerio 的答案中的参数映射到 SqlAlchemy 引擎：

from flask import json

engine = create_engine(
    app.config['SQLALCHEMY_DATABASE_URI'],
    convert_unicode=True,
    json_serializer=json.dumps,
)

您可以进一步扩展 FlaskJSONEncoder以支持decimal.Decimal以下内容：

import decimal

from flask import json

class CustomJSONEncoder(json.JSONEncoder):
    """
    Override Flask's `JSONEncoder.default`, which is called
    when the encoder doesn't handle a type.
    """
    def default(self, o):
        if isinstance(o, decimal.Decimal):
            return str(o)
        else:
            # raises TypeError: o not JSON serializable
            return json.JSONEncoder.default(self, o)

def init_json(app):
    """
    Use custom JSON encoder with Flask
    """
    app.json_encoder = CustomJSONEncoder