asyncpg
asyncpg copied to clipboard
Functions with record array paraments
I'm trying to invoke a function with a signature like this:
CREATE OR REPLACE FUNCTION "ns"."test"(p1 "ns"."my_type"[])
It results in the following exception being raised:
Traceback (most recent call last):
File "asyncpg/protocol/prepared_stmt.pyx", line 168, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
File "asyncpg/protocol/codecs/base.pyx", line 206, in asyncpg.protocol.protocol.Codec.encode
File "asyncpg/protocol/codecs/array.pyx", line 115, in asyncpg.protocol.protocol.array_encode
File "asyncpg/protocol/codecs/array.pyx", line 115, in asyncpg.protocol.protocol.array_encode
TypeError: a sized iterable container expected (got type 'dict')
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
buffer, _, self._exhausted = await protocol.bind_execute(
File "asyncpg/protocol/protocol.pyx", line 183, in bind_execute
File "asyncpg/protocol/prepared_stmt.pyx", line 197, in asyncpg.protocol.protocol.PreparedStatementState._encode_bind_msg
asyncpg.exceptions.DataError: invalid input for query argument $1: {'holiday_center': '123', 'observed_on':... (a sized iterable container expected (got type 'dict'))
How do I invoke such a function?
import asyncpg
async def invoke_function():
conn = await asyncpg.connect(user='your_user', password='your_password',
database='your_database', host='your_host')
# Define your input data as a list or array-like structure
input_data = [{"holiday_center": "123", "observed_on": "2022-08-29"}, {"holiday_center": "456", "observed_on": "2022-08-30"}]
try:
# Call the function with the input data
result = await conn.fetchval('SELECT "ns"."test"($1)', input_data)
# Handle the result as needed
print(result)
finally:
await conn.close()
# Run the asynchronous function
asyncio.get_event_loop().run_until_complete(invoke_function())