Cannot use custom codec for non-scalar type

• 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.

Lesen Sie asyncpg/connection.py im Umfeld von set_type_codec, und prüfen Sie anschließend asyncpg/protocol/settings.pyx sowie asyncpg/protocol/codecs/base.pyx, wo die Ablehnung von zusammengesetzten Typen erfolgt. Reproduzieren Sie den AgensGraph-Fall mit graphid/vertex/edge/graphpath und überprüfen Sie, dass für diese Typen benutzerdefinierte Codecs ohne den privaten Patch registriert werden können, während MATCH-Abfragen weiterhin funktionieren.