Most performant method to update many records?

[StayFoolisj]

I use Postgres 12 and a few times a day I need to update 30-80k rows and ideally in less than a minute.

Using SQLAlchemy / psycopg2 bulk_update_mappings is not really performant, its taking +10minutes even with chunking the data and using multithreading. So I started looking for options, which led me to find asyncpg.

Account model:

class Account(Base): __tablename__ = 'accounts' address = Column(String, primary_key=True) protocols_used = Column(ARRAY(String)) 

A snippet of what I'm doing now (minus chunking & multithreading for readability). My desired outcome is a significant speed improvement on this. 1k row updates per 1s would be ideal. I'm not sure what is realistic.

users_to_update_data = [acc for acc in all_accounts_data if acc['address'] in users_to_update] updates = [] for account_data in users_to_update_data: updates.append( { "address": account_data['address'], "protocols_used": account_data['protocols_used'] + [protocol] } ) session.bulk_update_mappings(Account, entries_to_update) 

From what I understand, this could be a job for executemany() ? I can prepare a list of either dicts or tuples (I believe tuples are more efficiently iterated due to their immutability?). I failed to find any code examples in the docs or elsewhere. Would be very grateful if someone could point me in the right direction with an example or suggestions. Thanks!

[elprans]

Indeed, use executemany:

updates = [(account_id, new_address, additional_protocol) from <data_source>] await connection.executemany( """  UPDATE accounts  SET   address = $2,  protocols_used = array_append(protocols_used, $3)  WHERE  id = $1  """, updates, )

[DoobyDouglas]

Awesome!

[StayFoolisj]

Thank you so much!

I refactored my script to using your suggestion + using sets instead of lists and total duration for my script now takes 21s instead of 20minutes! That involves fetching 500k records and updating 50k of them.

[elprans]

Did you use asyncpg to fetch records as well?

[StayFoolisj]

Yes but its a bit slower than the SQLAlchemy equivalent. Maybe because of the connection management?

async def get_all_accounts_asyncpg(): """ Returns all accounts in the database """ connection = await asyncpg.connect(**db_params) query = await connection.fetch('SELECT * FROM accounts') await connection.close() return [dict(r) for r in query] 

def get_all_accounts_sqlalchemy(): """ Returns all accounts in the database """ query = session.query(Account).all() return [r.address for r in query] 

[elprans]

You don't need to cast asyncpg.Record to dict, it's already dict-like. Remove the cast and see what happens.

[bobir01]

You don't need to cast asyncpg.Record to dict, it's already dict-like. Remove the cast and see what happens.

Good part of the day
I would like to construct data structure similar to asyncpg.Record for my project, is it possible with only pure paython ?
thanks