Asyncpg: Investigate the feasibility of implementing `executemany`

[styvane]

How to insert multiple rows without a traditional for loop?
Something like executemany in psycopg2

I am asking because I failed to find any example or method in the documentation.

Actually what I am doing is:

async def insertdemo(data, dns=DNS):
    async with asyncpg.create_pool(dns) as pool:
        async with pool.acquire() as con:
            async with con.transaction():
                stmt = '''insert into demo (num, name) select * from unnest($1::int[], $2::varchar[]);'''
                await con.execute(stmt, *zip(*data))

I would like to avoid to _unzip_ the data array.

[wrobell]

According to https://magic.io/blog/asyncpg-1m-rows-from-postgres-to-python/ asyncpg supports arrays and composite types, so I can imagine that determining semantics of value in con.excute(query, value) can be tricky.

However, it would be good to have at least executemany method, so

for row in data:
    await con.execute(query, *row)

can be executed as

await con.executemany(query, data)

where data could be any iterable.

BTW. How con.execute(query, *row) executed multiple times works exactly? Does it send each statement one by one over network or are multiple rows bundled into a batch?

[wrobell]

Also, looking at http://initd.org/psycopg/docs/usage.html

# Pass data to fill a query placeholders and let Psycopg perform
# the correct conversion (no more SQL injections!)
>>> cur.execute("INSERT INTO test (num, data) VALUES (%s, %s)",
...      (100, "abc'def"))

so asyncpg connection execute method seems to be incompatible with DB-API?

[wrobell]

What about changing execute method signature to

    async def execute(self, query: str, *args, data=None, timeout: float=None) -> str:

If args is non-empty then raise deprecation warning. If both args and data are specified, then raise TypeErrror. After few asyncpg versions, the args positional parameters could be removed.

[elprans]

asyncpg explicitly supports prepared statements, so there is no need for a dedicated executemany method. Simply iterate over a loop, calling the prepared statement:

ps = await conn.prepare('INSERT INTO test(num, data) VALUES($1, $2)')
for row in data:
   await ps.fetchval(*row)

Or, since the statements are prepared by default, you can use even simpler form:

for row in data:
    await conn.execute('INSERT INTO test(num, data) VALUES($1, $2)', *row)

[wrobell]

IMHO, dedicated executemany method would simplify API. It is easier to write

await conn.executemany('...', data)

than the whole loop. If this was implemented at Cython level, wouldn't it be faster as well?

I hope you do not mind, I will repeat the question from my first comment - are the rows iterated in the loop sent one by one or are the statements batched?

[elprans]

IMHO, dedicated executemany method would simplify API. It is easier to write

await conn.executemany('...', data)

than the whole loop. If this was implemented at Cython level, wouldn't it be faster as well?

No, high-level APIs in asyncpg are not Cythoned, and even if they were, the perf difference would be minimal.

I hope you do not mind, I will repeat the question from my first comment - are the rows iterated in the loop sent one by one or are the statements batched?

There is no such thing as batching, psycopg2 does the same row-by-row statement execution.

That said, event loop overhead needs to be benchmarked for this case. If it turns out to be significant, we will look into implementing the executemany method.

Meanwhile you can create a helper method using the loop technique mentioned above.