SQLAlchemy is Python's SQL toolkit and ORM, and in data engineering it's the layer that pipeline code uses to talk to relational databases without hand-concatenating query strings. It comes in two halves that you can use independently: the Core expression language, which builds SQL programmatically against table metadata, and the ORM, which maps rows to Python objects. For pipeline work - bulk loads, parameterized extracts, upserts - Core and the connection/engine machinery usually matter more than the object mapping.

Engines, connections, and parameterized SQL

Everything starts with an Engine, a factory and connection pool for a database URL. The engine manages connections so your pipeline doesn't open a fresh socket per query. Passing parameters through SQLAlchemy (rather than formatting them into the string) gives you correct quoting and protection against injection for free.

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg://etl:***@db.internal/warehouse",
                       pool_size=5, pool_pre_ping=True)

with engine.begin() as conn:  # begin() = transaction, auto-commit on exit
    rows = conn.execute(
        text("SELECT user_id, SUM(amount) AS revenue "
             "FROM orders WHERE day = :d GROUP BY user_id"),
        {"d": "2026-06-17"},
    ).mappings().all()

pool_pre_ping=True checks a pooled connection is still alive before handing it out - worth setting for long-lived ETL processes where the database may close idle connections.

The Core expression language

Core lets you describe a query as Python objects built from table metadata, so the statement composes and reflects rather than being a string you assemble by hand. This is what you want when query shape depends on runtime conditions, or when you want the same code to target Postgres and SQLite without dialect-specific string edits.

from sqlalchemy import MetaData, Table, select, func

md = MetaData()
orders = Table("orders", md, autoload_with=engine)  # reflect columns

stmt = (
    select(orders.c.user_id, func.sum(orders.c.amount).label("revenue"))
    .where(orders.c.status == "completed")
    .group_by(orders.c.user_id)
)

with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(row.user_id, row.revenue)

Because the query is an object, you can conditionally add a .where(...) or swap the aggregate without rebuilding text, and SQLAlchemy renders the right SQL for the connected dialect.

Loading data and chunked reads

For ingest, the fastest path is a multi-row insert - pass a list of dicts to execute and SQLAlchemy batches them. For upserts, use the dialect's insert (on_conflict_do_update on Postgres, on_duplicate_key_update on MySQL). When extracting large result sets, stream them so the whole table doesn't materialize in memory at once, which pairs well with handing each chunk to pandas for transformation.

# bulk insert: one round trip, many rows
with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO staging (user_id, revenue) VALUES (:user_id, :revenue)"),
        [{"user_id": r.user_id, "revenue": r.revenue} for r in rows],
    )

# stream a large extract instead of loading it whole
with engine.connect().execution_options(stream_results=True) as conn:
    for chunk in conn.execute(select(orders)).partitions(10_000):
        process(chunk)

The ORM and its Session are the right choice when you're modeling application entities with relationships and identity; for set-based pipeline operations, Core stays closer to the SQL the database actually runs.

Frequently asked questions

What's the difference between Core and the ORM? Core is the SQL expression language and engine machinery - you build and run statements that map directly to SQL. The ORM adds object-relational mapping, a session/identity map, and relationship loading on top of Core. Pipeline code often uses Core alone; application code more often uses the ORM.

Do I need an ORM for ETL? Usually not. ETL is mostly set-based bulk work where Core (or even raw text() through the engine) is clearer and faster than mapping rows to objects. The ORM shines for transactional application logic, not for moving millions of rows.

How do I avoid SQL injection? Never format user values into the query string. Use bound parameters - text("... :name") with a params dict, or Core's expression objects - and SQLAlchemy escapes and binds them correctly for the dialect.

Why use the connection pool instead of connecting per query? Opening a database connection is expensive; the engine's pool reuses connections across queries, which dramatically lowers latency and load for a process running many statements. Set pool_pre_ping=True so stale connections are detected and replaced.