JSONB в asyncpg: как получать объекты Python через set_type_codec

Почему asyncpg возвращает JSONB строками и как это исправить: зарегистрируйте кодек set_type_codec с json.dumps/json.loads, чтобы получать dict и list в Python.

При работе с asyncpg и JSONB в PostgreSQL нередко оказывается, что вместо нативных объектов Python вы получаете обычные строки. Если запросы возвращают значения JSONB как текст, операции, рассчитывающие на dict и list, становятся неудобными, а предположения о типах мгновенно ломаются. Ниже — краткое объяснение проблемы и минимальное, надежное решение.

Как воспроизвести проблему

Ниже — минимальный скрипт: он выбирает литерал JSONB и выводит результат. По выводу видно, что значение приходит как строка.

import asyncio
import asyncpg

async def runner():
    link = await asyncpg.connect("postgres://user:pass@localhost/database")
    rows = await link.fetch("select '[1, 2, 3]'::jsonb as col;")
    for rec in rows:
        for k, v in rec.items():
            print("'" + k + "'")
            print("'" + v + "'")

if __name__ == "__main__":
    asyncio.run(runner())

Ожидался список, но в выводе — строка в кавычках. Интроспекция подтверждает, что тип значения — str.

Что происходит и почему

Чтобы asyncpg автоматически преобразовывал JSONB в нативные типы Python, нужно зарегистрировать кодек для jsonb. Без явного кодека в таком случае значение возвращается как текст. Такое поведение соответствует рекомендациям из обсуждений в сообществе и примеру самой библиотеки по автоматическому преобразованию JSON, где показано, что для jsonb требуется вызвать set_type_codec.

Чтобы asyncpg автоматически декодировал значения jsonb в объекты Python, необходимо явно зарегистрировать пользовательский кодек типов.

Решение: зарегистрируйте кодек для jsonb

Регистрация кодека сообщает asyncpg, как сериализовать и десериализовать JSONB. Пары json.dumps и json.loads достаточно для корректного обмена между PostgreSQL и структурами данных Python.

import asyncio
import asyncpg
import json

async def run_fixed():
    dbh = await asyncpg.connect("postgres://user:pass@localhost/database")

    await dbh.set_type_codec(
        'jsonb',
        encoder=json.dumps,
        decoder=json.loads,
        schema='pg_catalog'
    )

    result = await dbh.fetch("select '[1, 2, 3]'::jsonb as col;")
    for rec in result:
        parsed = rec['col']
        print(parsed, type(parsed))  # Результат: [1, 2, 3] <class 'list'>

    await dbh.close()

if __name__ == "__main__":
    asyncio.run(run_fixed())

После добавления set_type_codec значения JSONB приходят как полноценные dict и list, поэтому остальной код может работать с ними напрямую.

Почему это важно

Единообразные типы критичны при построении пайплайнов данных, написании валидации и формировании запросов, которые питают приложение. Если JSONB приходит строками, приходится парсить его на лету, возрастает риск скрытых ошибок, а конверсия рассеивается по коду. Централизация поведения на уровне соединения делает модель данных предсказуемой и упрощает код.

Выводы

Если рассчитываете получать нативные объекты Python для JSONB в asyncpg, явно зарегистрируйте кодек jsonb через set_type_codec с json.dumps и json.loads. Рано проверяйте получаемые типы и держите политику преобразования рядом с настройкой соединения. Этот небольшой шаг предотвращает несоответствия типов и делает границу ввода‑вывода чище.