Cannot use custom codec for non-scalar type

[OndrejZapletal]

• asyncpg version: 0.18.3
• PostgreSQL version: AgensGraph 2.1.0, based on PostgreSQL 10.4
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce the issue with a local PostgreSQL install?: Does not apply
• Python version: 3.7.2
• Platform: Ubuntu 16.04.3 LTS
• Do you use pgbouncer?: No
• Did you install asyncpg with pip?: Yes
• If you built asyncpg locally, which version of Cython did you use?: Not relevant
• Can the issue be reproduced under both asyncio and uvloop?: Not relevant

I'm trying to use asyncpg for communication with AgensGraph. It is graph database build on top of PostgreSQL 10.4. Bitnine provides client agensgraph-python for it, but is built on top of psycopg2 and therefore dosn't support async. I've tried to simply use asyncpg instead with following simple example:

import asyncio
import asyncpg

async def run():
    conn = await asyncpg.connect(
        database = 'agens',
        user = 'agens',
        host = '127.0.0.1',
        port = 5435
    )

    await conn.execute('''
        DROP GRAPH IF EXISTS t CASCADE;
        CREATE GRAPH t;
        SET graph_path = t;
        CREATE (:v {name: 'AgensGraph'});
    ''')

    values = await conn.fetch("MATCH (n) RETURN n")

    print(values)
    await conn.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

got following exception:

Traceback (most recent call last):
  File "example_script.py", line 56, in <module>
    loop.run_until_complete(run())
  File "/home/ondrej_zapletal/.pyenv/versions/3.7.2/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "example_script.py", line 50, in run
    values = await conn.fetch("MATCH (n) RETURN n")
  File "/home/ondrej_zapletal/.pyenv/versions/squid/lib/python3.7/site-packages/asyncpg/connection.py", line 421, in fetch
    return await self._execute(query, args, 0, timeout)
  File "/home/ondrej_zapletal/.pyenv/versions/squid/lib/python3.7/site-packages/asyncpg/connection.py", line 1414, in _execute
    query, args, limit, timeout, return_status=return_status)
  File "/home/ondrej_zapletal/.pyenv/versions/squid/lib/python3.7/site-packages/asyncpg/connection.py", line 1422, in __execute
    return await self._do_execute(query, executor, timeout)
  File "/home/ondrej_zapletal/.pyenv/versions/squid/lib/python3.7/site-packages/asyncpg/connection.py", line 1434, in _do_execute
    stmt = await self._get_statement(query, None)
  File "/home/ondrej_zapletal/.pyenv/versions/squid/lib/python3.7/site-packages/asyncpg/connection.py", line 341, in _get_statement
    settings.register_data_types(types)
  File "asyncpg/protocol/settings.pyx", line 35, in asyncpg.protocol.protocol.ConnectionSettings.register_data_types
  File "asyncpg/protocol/settings.pyx", line 36, in asyncpg.protocol.protocol.ConnectionSettings.register_data_types
  File "asyncpg/protocol/codecs/base.pyx", line 563, in asyncpg.protocol.protocol.DataCodecConfig.add_types
  File "asyncpg/protocol/codecs/base.pyx", line 677, in asyncpg.protocol.protocol.DataCodecConfig.declare_fallback_codec
NotImplementedError: unhandled standard data type 'graphid' (OID 7002)

I've looked at agensgraph-python implementation and it seems fairly straight forward. It simply adds new types [GRAPHID, VERTEX, EDGE, GRAPHPATH]. Like this:

https://github.com/bitnine-oss/agensgraph-python/blob/e6b9fedfda81e28a46ac9dcf8b8e562ca1b1119d/agensgraph/__init__.py#L26-L42

I've tried to use set_type_codec to define those, but I've come across this error:

https://github.com/MagicStack/asyncpg/blob/9d8a1c03597970f231da67415c6465776adc03b7/asyncpg/connection.py#L968-L971

My question is why it isn't allowed to add codecs for composite types?

I've overcame this problem temporarily by patching set_type_codec in following way:

import asyncio
import asyncpg

async def set_type_codec(conn, typenames):
    schema='pg_catalog'
    format='text'
    conn._check_open()
    for typename in typenames:
        typeinfo = await conn.fetchrow(
            asyncpg.introspection.TYPE_BY_NAME, typename, schema)
        if not typeinfo:
            raise ValueError('unknown type: {}.{}'.format(schema, typename))

        oid = typeinfo['oid']
        conn._protocol.get_settings().add_python_codec(
            oid, typename, schema, 'scalar',
            lambda a: a, lambda a: a, format)

    # Statement cache is no longer valid due to codec changes.
    conn._drop_local_statement_cache()

async def run():
    conn = await asyncpg.connect(
        database = 'agens',
        user = 'agens',
        host = '127.0.0.1',
        port = 5435
    )

    await set_type_codec(conn, ['graphid', 'vertex', 'edge', 'graphpath'])
    await conn.execute('''
        DROP GRAPH IF EXISTS t CASCADE;
        CREATE GRAPH t;
        SET graph_path = t;
        CREATE (:v {name: 'AgensGraph'});
        CREATE (:v {name: 'OtherData'});
    ''')

    values = await conn.fetch("MATCH (n) RETURN n")

    print(values)
    await conn.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

This seems to work for me so far, but I'm not sure if I won't come across some additional issues due to this in the future.

[elprans]

Composite types are normally handled automatically, since their (binary) structure is well defined. The issue in this case is that AgensGraph chose to put their extension types into the standard Postgres type namespace, and asyncpg expects to have explicit codecs for all such types. The fix here would be to handle "standard" composite types like all other composites.

[OndrejZapletal]

Thanks for the response. Could you elaborate a little. You mean fix for asyncpg? Or is there something I can do on my end instead. I would like to know if I can run into some problems in the future when I've decided to circumvent the asyncpg limit of codec addition for composite types?

[elprans]

To clarify: this can be fixed in asyncpg.

[davidandreoletti]

To clarify: this can be fixed in asyncpg.

Is there a plan in place at the moment ?

[larissaleite]

Any updates or workarounds on this?