</>
pythonintermediate

SQLAlchemy Bulk Insert with Upsert

Efficiently bulk-insert rows with conflict resolution using SQLAlchemy Core and PostgreSQL.

python
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.dialects.postgresql import insert

engine = create_engine('postgresql://user:pass@localhost/db')
meta = MetaData()
meta.reflect(bind=engine)
table = meta.tables['events']

rows = [
    {'id': 1, 'name': 'Alice', 'score': 95},
    {'id': 2, 'name': 'Bob',   'score': 87},
]

stmt = insert(table).values(rows)
stmt = stmt.on_conflict_do_update(
    index_elements=['id'],
    set_={'score': stmt.excluded.score},
)
with engine.begin() as conn:
    conn.execute(stmt)
    print('Upserted', len(rows), 'rows')

Use Cases

  • idempotent loads
  • incremental ETL
  • data synchronisation

Tags

#sqlalchemy#postgres#upsert#bulk

Related Snippets

Similar patterns you can reuse in the same workflow.

pythonadvanced

SQLAlchemy Async Session with asyncpg

Use SQLAlchemy 2.0 async sessions with asyncpg for non-blocking database access in async pipelines.

Best for: async web services

#sqlalchemy#async
sqlbeginner

PostgreSQL COPY — Fast CSV Import

Use PostgreSQL COPY command for high-speed bulk data loading from CSV files with error handling.

Best for: High-speed bulk data loading into PostgreSQL

#postgres#copy
bashbeginner

Database Backup and Restore to S3

Automated PostgreSQL backup script with compression, S3 upload, retention policy, and restore commands.

Best for: Automated daily database backups to S3

#bash#backup
pythonadvanced

Delta Lake MERGE with Python

Perform ACID upserts on a Delta Lake table using the delta-rs Python binding.

Best for: lakehouse upserts

#delta-lake#upsert