TopolearnImports, syncs and ETL jobs need high-throughput writes and insert-or-update semantics that per-object add() can't deliver.
Adding thousands of ORM objects with session.add_all() works, but each object carries identity-map and event overhead. For bulk loads, pass a list of dicts to an insert statement: session.execute(insert(User), [{"name": "a"}, {"name": "b"}, ...]). SQLAlchemy batches it efficiently, using "insertmanyvalues" to send multi-row INSERTs, and still applies Python-side column defaults.
Add .returning(User.id) to get generated keys back, or insert(User).returning(User) to get ORM objects, on databases supporting RETURNING (PostgreSQL, SQLite, MariaDB, SQL Server).
Upserts are dialect-specific, so import insert from the dialect module:
- PostgreSQL and SQLite:
insert(...).on_conflict_do_nothing(index_elements=[...])or.on_conflict_do_update(index_elements=["email"], set_={"name": stmt.excluded.name}).excludedrefers to the row you tried to insert. - MySQL and MariaDB:
.on_duplicate_key_update(name=stmt.inserted.name).
For bulk updates by primary key, session.execute(update(User), [{"id": 1, "name": "x"}, ...]) performs an executemany of UPDATE statements.
Batch very large loads into chunks of a few thousand rows per statement and commit periodically so a single transaction doesn't grow unbounded. Bulk statements skip ORM events and relationship cascades.
Example
from sqlalchemy import create_engine, select, insert
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase): pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
email: Mapped[str] = mapped_column(unique=True)
name: Mapped[str]
engine = create_engine("sqlite://"); Base.metadata.create_all(engine)
with Session(engine) as s:
rows = [{"email": f"u{i}@x.io", "name": f"user{i}"} for i in range(1000)]
ids = s.scalars(insert(User).returning(User.id), rows).all()
print(len(ids), ids[:3]) # 1000 [1, 2, 3]
stmt = sqlite_insert(User).values(email="u0@x.io", name="Renamed")
stmt = stmt.on_conflict_do_update(index_elements=["email"],
set_={"name": stmt.excluded.name})
s.execute(stmt); s.commit()
print(s.scalar(select(User.name).where(User.id == 1))) # RenamedRemember
- session.execute(insert(Model), list_of_dicts) for fast bulk inserts
- .returning() gets generated ids back in the same round trip
- Upserts use the dialect's insert: on_conflict_do_update / on_duplicate_key_update
- Chunk huge loads and commit periodically
Watch out for
- Using the generic sqlalchemy.insert and looking for on_conflict_do_update, which only exists on dialect inserts.