SQLAlchemy 2.x for Data Engineers: Core, ORM, Bulk Inserts, Async Engines

sqlalchemy 2 data engineers is the Python toolkit every data engineer eventually reaches for when the pipeline needs to move beyond raw psycopg2 cursors and start writing typed, connection-pooled, dialect-portable database code — and SQLAlchemy 2.x's release brought a significant rewrite that unified the Core and ORM APIs, added first-class async support, and made typed models via Mapped[T] the default. Every Python DE eventually writes SQLAlchemy code; knowing when to use Core versus ORM, how to batch inserts without the "1M single-row INSERT" perf disaster, and when to reach for asyncpg behind an AsyncSession is what separates a mid-level Python engineer from a senior one. This guide walks the 2.x-era patterns that actually matter for pipeline code.

The tour walks the five pillars — (1) the Core vs ORM decision matrix (Core for ETL and analytical writes, ORM for API layers with relationship graphs), (2) bulk insert patterns including executemany, insert().values(), and Postgres-specific INSERT ... ON CONFLICT DO UPDATE via dialect.insert(), (3) async engines with create_async_engine, AsyncSession, and asyncpg / aiomysql drivers, (4) connection pooling strategies covering QueuePool for long-lived services, NullPool for AWS Lambda, pool_pre_ping for surviving stale connections through NAT / load balancer timeouts, and (5) the dialect matrix showing how Postgres, MySQL, SQL Server, Oracle, Snowflake, BigQuery, Databricks, and SQLite are supported. Every section ships a Solution-Tail interview answer — code, trace, output, why-this-works.

Practice on SQL library →, optimization →, and joins →.

---

On this page

• Why SQLAlchemy 2.x matters
• Core vs ORM decision
• Bulk insert and upsert patterns
• Async engines with asyncpg
• Connection pooling and dialect matrix
• Cheat sheet
• FAQ
• Practice on PipeCode

---

1. Why SQLAlchemy 2.x matters in 2026

sqlalchemy 2 data engineers — the toolkit that unified Core + ORM under one typed API

The one-sentence invariant: SQLAlchemy 2.x consolidated the Core and ORM APIs under a single select() / Session model, replaced the legacy Query object with session.execute(select(Model)), brought typed models via Mapped[T] (PEP 484 friendly), and added first-class async support via create_async_engine + AsyncSession — for data engineers, this means one Python DB library that scales from single-row API queries to bulk ETL to async streaming pipelines without rewriting the data-access layer.

Where SQLAlchemy shows up in DE pipelines.

• ETL scripts. Read from S3, transform in Pandas / Polars, write to Postgres / Snowflake via SQLAlchemy Core.
• API layers. FastAPI + SQLAlchemy ORM for CRUD endpoints with automatic relationship loading.
• dbt post-processors. Python scripts that read dbt models and write derived tables.
• CDC apply-loops. Consume Kafka, apply MERGE via SQLAlchemy Core insert().on_conflict_do_update().
• Data quality tests. SQLAlchemy Core queries against warehouse tables assert row counts and invariants.
• Airflow operators. PostgresHook wraps SQLAlchemy engine internally.
• Migration scripts. Alembic + SQLAlchemy for schema changes.

What senior interviewers probe.

• Core vs ORM decision. When to use each — Core for bulk ETL, ORM for relationship graphs.
• Bulk insert. executemany vs insert().values() vs session.bulk_insert_mappings.
• Connection pool. QueuePool vs NullPool for Lambda; pool_pre_ping for NAT gateway timeout.
• Async. AsyncEngine + AsyncSession for I/O-bound pipelines.
• Typed models. Mapped[int], Mapped[str], mapped_column().
• Session scope. Per-request in web apps; per-batch in ETL; never long-lived across transactions.

Worked example — the 100× bulk insert speedup

# Bad: 1M individual INSERTs (30 minutes)
for row in rows:
    session.execute(insert(User).values(**row))
session.commit()

# Good: batched executemany (2 minutes)
session.execute(insert(User), rows)
session.commit()

# Best: Postgres COPY via raw connection (20 seconds)
with engine.raw_connection() as conn:
    with conn.cursor() as cur:
        with cur.copy("COPY users (id, name, email) FROM STDIN") as copy:
            for row in rows:
                copy.write_row((row['id'], row['name'], row['email']))

Rule of thumb. For > 10K rows to Postgres, use COPY. For 100-10K, use insert().values(list_of_dicts).

---

2. Core vs ORM decision

The Core–ORM decision matrix — when to use each API surface

The mental model in one line: SQLAlchemy Core exposes Tables + Column + SQL expression language — you construct SQL programmatically and execute it; the ORM adds Mapped classes with declarative relationships, session-managed unit-of-work, and identity map — Core is closer to the metal and cheaper per row; ORM is easier to model complex object graphs but pays overhead per row.

Slot 1 — Core basics.

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select, insert

engine = create_engine("postgresql://user:pass@host/db")
metadata = MetaData()
users = Table("users", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(100)),
    Column("email", String(200)),
)
metadata.create_all(engine)

with engine.connect() as conn:
    result = conn.execute(select(users).where(users.c.id == 42))
    for row in result:
        print(row.name)
    conn.commit()

Slot 2 — ORM basics (2.x typed).

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)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200))

with Session(engine) as session:
    user = session.execute(select(User).where(User.id == 42)).scalar_one()
    print(user.name)

Slot 3 — decision matrix.

Aspect	Core	ORM
Cost per row	Low	Higher (session tracking)
Type safety	Column-level	Full model-level
Relationships	Manual JOIN	Declarative relationship()
Best for	ETL, bulk, analytical	API, CRUD, complex graphs

Slot 4 — Core wins.

• Bulk inserts / updates.
• Reading millions of rows for aggregation.
• Dialect-portable SQL construction.
• Data validation pipelines.

Slot 5 — ORM wins.

• API endpoints with nested JSON responses.
• Complex object graphs with lazy loading.
• Transactional business logic.
• When the model IS the domain.

Slot 6 — hybrid pattern.

# ORM for the write; Core for the analytical read
with Session(engine) as sess:
    sess.add_all([User(name=f"user{i}") for i in range(100)])
    sess.commit()

with engine.connect() as conn:
    stats = conn.execute(select(func.count(users.c.id))).scalar_one()

Use each where it fits.

---

3. Bulk insert and upsert patterns

High-throughput write patterns — executemany, insert().values(), and dialect-specific ON CONFLICT

The mental model in one line: single-row INSERT per statement is 100–1000× slower than batched writes; insert(Table).values(list_of_dicts) batches into a single multi-row VALUES statement, executemany batches into a driver-level parameter array, dialect.insert() gives you Postgres ON CONFLICT and MySQL ON DUPLICATE KEY UPDATE, and for the biggest loads, drop to COPY (Postgres) or LOAD DATA (MySQL) via the raw driver.

Slot 1 — executemany style.

from sqlalchemy import insert
rows = [{"id": i, "name": f"user{i}"} for i in range(10_000)]
with engine.begin() as conn:
    conn.execute(insert(users), rows)  # SQLAlchemy 2.x: pass list as second arg

Slot 2 — multi-row VALUES.

from sqlalchemy import insert
stmt = insert(users).values([
    {"id": 1, "name": "a"},
    {"id": 2, "name": "b"},
])
conn.execute(stmt)

Slot 3 — Postgres ON CONFLICT.

from sqlalchemy.dialects.postgresql import insert as pg_insert

stmt = pg_insert(users).values(id=42, name="alice")
stmt = stmt.on_conflict_do_update(
    index_elements=["id"],
    set_={"name": stmt.excluded.name}
)
conn.execute(stmt)

Slot 4 — MySQL ON DUPLICATE KEY.

from sqlalchemy.dialects.mysql import insert as mysql_insert

stmt = mysql_insert(users).values(id=42, name="alice")
stmt = stmt.on_duplicate_key_update(name=stmt.inserted.name)
conn.execute(stmt)

Slot 5 — COPY for max throughput on Postgres.

with engine.raw_connection() as conn:
    with conn.cursor() as cur:
        with cur.copy("COPY users (id, name) FROM STDIN") as copy:
            for row in rows:
                copy.write_row((row['id'], row['name']))
    conn.commit()

10-100× faster than any INSERT approach on Postgres.

Slot 6 — bulk_insert_mappings (ORM).

session.bulk_insert_mappings(User, list_of_dicts)
session.commit()

Skips ORM overhead; comparable to Core executemany.

Slot 7 — throughput comparison.

Method	1M rows
Single INSERT per row	30 minutes
executemany	3 minutes
Multi-row VALUES	2 minutes
Postgres COPY	20 seconds

---

4. Async engines with asyncpg

sqlalchemy async — the async pattern for I/O-bound pipelines

The mental model in one line: create_async_engine("postgresql+asyncpg://...") returns an AsyncEngine; use AsyncSession in async with blocks; every session.execute becomes await session.execute — the ceremony is small but the throughput win on I/O-heavy pipelines is 10-100× because you can fan out queries in parallel.

Slot 1 — basic setup.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")

async def get_user(user_id: int):
    async with AsyncSession(engine) as sess:
        result = await sess.execute(select(User).where(User.id == user_id))
        return result.scalar_one()

Slot 2 — fan-out with asyncio.gather.

import asyncio

async def fetch_all():
    async with AsyncSession(engine) as sess:
        tasks = [
            sess.execute(select(User).where(User.id == i))
            for i in range(100)
        ]
        results = await asyncio.gather(*tasks)

Slot 3 — drivers.

• asyncpg — best-in-class Postgres async driver. Fastest.
• aiopg — legacy; use asyncpg instead.
• aiomysql — MySQL async.
• aioodbc — SQL Server via ODBC (slower).

Slot 4 — session per request.

# FastAPI pattern
from fastapi import Depends

async def get_session():
    async with AsyncSession(engine) as sess:
        yield sess

@app.get("/users/{user_id}")
async def read_user(user_id: int, sess=Depends(get_session)):
    return await sess.execute(select(User).where(User.id == user_id)).scalar_one()

Slot 5 — bulk async writes.

async with AsyncSession(engine) as sess:
    async with sess.begin():
        sess.add_all([User(name=f"u{i}") for i in range(10_000)])
    # commit on context exit

Slot 6 — gotchas.

• Never mix sync and async engines in same code path.
• AsyncSession transactions differ subtly from sync Session.
• Lazy-loading on ORM relationships requires await — use selectinload for eager loading.

---

5. Connection pooling and dialect matrix

Connection pool strategies — QueuePool for services, NullPool for Lambda, pool_pre_ping for stale connections

The mental model in one line: the connection pool is what keeps your app fast (avoids TCP + auth handshake per query) but also what breaks when NAT / load balancer / firewall idles-out connections that Python didn't know were dead — pool_pre_ping=True sends a lightweight query before each checkout, pool_recycle=1800 forces recycle after 30 minutes, NullPool disables pooling entirely for AWS Lambda where connections can't survive execution boundaries.

Slot 1 — QueuePool (default).

engine = create_engine("postgresql://...",
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
    pool_pre_ping=True,
    pool_recycle=1800,
)

Slot 2 — NullPool for Lambda.

from sqlalchemy.pool import NullPool
engine = create_engine("postgresql://...", poolclass=NullPool)

Every request opens fresh connection; no cross-invocation state.

Slot 3 — pool_pre_ping.

Sends SELECT 1 before checkout. Detects dead connections silently killed by NAT / firewall / load balancer. Cheap; enable everywhere.

Slot 4 — pool_recycle.

Forces connections to be recycled after N seconds. Combines with pool_pre_ping for defense in depth.

Slot 5 — 8-engine driver matrix.

Engine	Sync driver	Async driver
Postgres	psycopg2 / psycopg3	asyncpg
MySQL	pymysql / mysqlclient	aiomysql
SQL Server	pyodbc	aioodbc
Oracle	cx_Oracle / oracledb	oracledb (async since 2.0)
SQLite	built-in	aiosqlite
Snowflake	snowflake-connector-python	not officially async
BigQuery	google-cloud-bigquery	native async client
Databricks	databricks-sql-connector	not async

Slot 6 — production checklist.

• pool_pre_ping=True — always.
• pool_recycle=1800 — 30 min < any NAT timeout.
• pool_size = expected concurrent workers.
• max_overflow = burst headroom.
• Never leak sessions — always with context.

---

Cheat sheet — SQLAlchemy 2.x recipe list

• Engine — create_engine("postgresql://...", pool_pre_ping=True).
• AsyncEngine — create_async_engine("postgresql+asyncpg://...").
• Core select — select(Table).where(col == val).
• ORM select — select(Model).where(Model.id == val).
• Bulk insert — conn.execute(insert(Table), list_of_dicts).
• Postgres UPSERT — pg_insert().values(...).on_conflict_do_update(index_elements=[...], set_={...}).
• MySQL UPSERT — mysql_insert().on_duplicate_key_update().
• COPY — raw connection cursor + COPY ... FROM STDIN.
• Session — per-request or per-batch; never long-lived.
• AsyncSession — same pattern with async with + await.
• Mapped types — Mapped[int], Mapped[str] = mapped_column(String(100)).
• Relationships — Mapped[List["Order"]] = relationship(back_populates="user").
• Eager load — .options(selectinload(User.orders)).
• Pool — QueuePool default; NullPool for Lambda.
• pre_ping — always True for prod services.
• recycle — 30 min for services behind NAT.
• Dialect — imported per engine for engine-specific features.

Frequently asked questions

When should I use Core vs ORM?

Use Core for bulk ETL, analytical queries returning many rows, and dialect-portable SQL construction where you build queries programmatically. Use ORM for CRUD API endpoints where you want an object graph with automatic relationship loading, and for complex business-logic transactions where the model represents domain state. Hybrid is fine — Core for writes and reads that don't need object mapping; ORM for endpoints that expose Python objects.

How do I make bulk inserts fast?

For Postgres, use COPY ... FROM STDIN via raw connection — 10-100× faster than any INSERT variant. For 10K-1M rows, use conn.execute(insert(Table), list_of_dicts) which internally batches via executemany. For smaller batches, insert().values(list) produces a single multi-row VALUES statement. Never loop single-row INSERTs in application code.

What's the difference between pool_pre_ping and pool_recycle?

pool_pre_ping sends a lightweight query (SELECT 1) before each connection checkout to verify the connection is alive; it catches connections killed silently by NAT, load balancer, or firewall idle timeout. pool_recycle forces connections to be closed and reopened after N seconds regardless of activity; use in combination — pre_ping catches dead connections proactively; recycle prevents them from getting old enough to die. In production, set both.

Why should I use NullPool in AWS Lambda?

Lambda functions can be killed between invocations; connections held in a pool are lost. QueuePool assumes long-lived process; NullPool creates fresh connection per operation. For Lambda, use NullPool + a proxy like RDS Proxy or PgBouncer that pools upstream.

How does async work with SQLAlchemy?

create_async_engine returns an AsyncEngine; wrap it with AsyncSession; every execute becomes await execute. The pattern is the same as sync but with async context managers. Use asyncpg driver (Postgres) or aiomysql (MySQL). Best for I/O-bound pipelines where you fan out many queries in parallel via asyncio.gather.

Should I still use Query() or is select() the way?

In 2.x, always use select(). The legacy Query API is deprecated and less flexible; select() works identically in Core and ORM, supports async, and is what all documentation examples now show.

Practice on PipeCode

• SQL practice library →
• SQL optimization drills →
• SQL join drills →
• SQL for DE course →

Pipecode.ai is Leetcode for Data Engineering — every `sqlalchemy 2 data engineers` pattern above ships with hands-on practice rooms where you switch between Core and ORM, benchmark bulk insert strategies, wire up asyncpg + AsyncSession, tune pool_pre_ping for NAT survival, and stitch Alembic migrations into CI — the exact Python + SQL fluency that senior DE interviews probe.

Practice SQL now →
Optimization drills →