backend dialect json serializer/deserializer setting?

[jaddison]

Is there a (much!) better way to achieve the following? (Note the hack at the bottom of the snippet)

def json_loads(v):
    return json.loads(v)


def serialize_unknown_types(v):
    if isinstance(v, Decimal):
        return str(v)
    raise TypeError


def json_dumps(v, *, default=serialize_unknown_types):
    return json.dumps(v, default=default)

db = databases.Database(settings.PG_DSN, min_size=settings.PG_POOL_MIN, max_size=settings.PG_POOL_MAX, init=init)

# HACCCCCKKKKYYYY
db._backend._dialect._json_serializer = json_dumps
db._backend._dialect._json_deserializer = json_loads

I needed this for Decimal values inside a dict that was being saved into a PostgreSQL JSONB field.

[Vlad97vsv]

You can use the init method for asyncpg. Works great for me

from asyncpg import Connection
from databases import Database as BaseDatabase
import orjson

from app.utils.meta import MetaSingleton


async def _init_json_codecs(connection: Connection):
    await connection.set_type_codec(
        "json",
        encoder=orjson.dumps,
        decoder=orjson.loads,
        schema="pg_catalog",
    )


class Database(BaseDatabase, metaclass=MetaSingleton):
    def __init__(self) -> None:
        pass

    async def init(self, *args, **kwargs) -> None:
        super().__init__(*args, init=_init_json_codecs, **kwargs)
        await self.connect()