SQLAlchemy for Web Scraping: A Production Storage Layer

My short answer: in a scraping pipeline SQLAlchemy is not the part that fetches — it is the part that makes fetching again unnecessary. Model the record you actually need, bound the connection pool below the database's own limit, write every row through an ON CONFLICT DO UPDATE so a retry costs nothing, and retry the whole transaction rather than the statement. Everything below targets SQLAlchemy 2.0 — 2.0.52, released 11 August 2026 — and states the official defaults rather than guessing at them.

Where the database sits in a scraping pipeline

A production pipeline has five stages — URL sourcing, fetching, parsing, validation, and storage — and our web scraping guide walks the whole chain. Only one of those stages costs money per byte. Residential proxy traffic is billed per gigabyte, so the economics of the pipeline are decided by how often you have to fetch a page a second time; our web scraping cost estimation breakdown puts real numbers on that.

That reframes the storage layer. Its job is not "keep the rows somewhere". Its job is to answer three questions cheaply enough that the fetcher never has to:

1. Have I already seen this record, and has it changed? If the answer is no, the crawl skips a paid request.
2. What survives a crash? A worker that dies mid-batch must not leave half a page committed and half lost.
3. Can analysts read while the crawler writes? A long-running transaction that holds locks makes the warehouse and the crawler enemies.

SQLAlchemy answers all three, in two layers that are worth keeping straight. Core is the SQL expression language: Table, select(), insert(), the engine and the connection pool. The ORM adds mapped classes, an identity map and the unit-of-work Session on top of it. The productive split for a scraper is to define the schema with ORM declarative models — because the Python types and the DDL then have one source of truth — and to write batches with Core-style statements, because inserting ten thousand rows is a set operation, not ten thousand object mutations.

Model the record, not the page you scraped

The most expensive modelling mistake in a scraper is storing the artefact instead of the fact. Raw HTML is enormous, changes for reasons you do not care about, and cannot be queried. Store the extracted record, plus enough metadata to decide whether it is new.

from __future__ import annotations

import datetime as dt
from typing import Annotated, Any

from sqlalchemy import (
    BigInteger,
    ForeignKey,
    Index,
    MetaData,
    String,
    TIMESTAMP,
    UniqueConstraint,
    func,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

# Column recipes declared once and annotated everywhere. Changing the policy
# for every timestamp in the schema is then a one-line edit, not a sweep.
bigint_pk = Annotated[int, mapped_column(BigInteger, primary_key=True)]
utc_now = Annotated[dt.datetime, mapped_column(server_default=func.now())]

# Every constraint gets a deterministic name. Alembic cannot generate a
# migration for a constraint the database named anonymously, so this is what
# makes autogenerate usable later.
NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}


class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING_CONVENTION)

    # A naive timestamp is the classic silent data loss in a crawler that runs
    # in more than one region. Fix it once, for every model in the schema.
    type_annotation_map = {
        dt.datetime: TIMESTAMP(timezone=True),
        dict[str, Any]: JSONB,
    }


class ScrapedPage(Base):
    __tablename__ = "scraped_page"

    id: Mapped[bigint_pk]
    source: Mapped[str] = mapped_column(String(64))
    external_id: Mapped[str] = mapped_column(String(256))
    url: Mapped[str] = mapped_column(String(2048))
    # Mapped[str] is NOT NULL; Mapped[str | None] is NULL. The annotation IS
    # the schema, so mypy and the DDL cannot disagree about optionality.
    title: Mapped[str | None]
    payload: Mapped[dict[str, Any]]
    # Hash what you EXTRACTED, never the raw HTML: hash the page and every
    # rotating ad slot and CSRF token reads as a content change.
    content_hash: Mapped[str] = mapped_column(String(64))
    first_seen_at: Mapped[utc_now]
    last_changed_at: Mapped[utc_now]

    observations: Mapped[list[PriceObservation]] = relationship(
        back_populates="page", cascade="all, delete-orphan"
    )

    __table_args__ = (
        # The natural key: what "the same page, seen again" means. Without it
        # the upsert in the next section has nothing to conflict on.
        UniqueConstraint("source", "external_id", name="uq_scraped_page_natural_key"),
        Index("ix_scraped_page_changed", "source", "last_changed_at"),
    )


class PriceObservation(Base):
    __tablename__ = "price_observation"

    id: Mapped[bigint_pk]
    page_id: Mapped[int] = mapped_column(ForeignKey("scraped_page.id", ondelete="CASCADE"))
    observed_at: Mapped[utc_now]
    # Money as integer minor units. A float price silently loses cents, and
    # you will not notice until someone reconciles a year of history.
    amount_minor: Mapped[int]
    currency: Mapped[str] = mapped_column(String(3))

    page: Mapped[ScrapedPage] = relationship(back_populates="observations")

Three details in there are doing most of the work.

Mapped[] decides nullability. The documented rule is that Mapped[str] produces NOT NULL and Mapped[Optional[str]] — or Mapped[str | None] — produces NULL, with primary_key=True implying NOT NULL regardless. You can still override it with mapped_column(nullable=...), but the default means your type checker is reading the same truth as your schema.

type_annotation_map centralises the Python-to-SQL mapping. SQLAlchemy already maps int to Integer, str to String, datetime.datetime to DateTime, uuid.UUID to Uuid and so on. Overriding the map on the base class is how you apply a project-wide policy — timezone-aware timestamps, JSONB instead of generic JSON — without repeating it on 40 columns.

The natural key is a real constraint, not a convention. source plus external_id is what makes the same product page recognisable across crawls. A surrogate id alone cannot express it, and everything in the next section depends on the database being able to detect the collision itself.

The engine and the pool are your concurrency budget

create_engine() is a factory, not a connection: it holds a connection pool that is created lazily and shared. For anything other than SQLite in-memory, the default pool is QueuePool, and its documented defaults are the numbers that will bite you first.

The arithmetic that matters is workers × (pool_size + max_overflow) ≤ server connection limit − headroom. With the defaults, eight crawler processes can demand 120 connections from a database that may allow 100 in total, and the failure shows up as pool_timeout expiring in the workers while a migration cannot connect at all. Decide the ceiling deliberately:

import os

from sqlalchemy import create_engine

engine = create_engine(
    # Never a literal DSN. Credentials belong in the environment or a secrets
    # manager, and the URL carries the driver: postgresql+psycopg for sync.
    os.environ["SCRAPER_DATABASE_URL"],
    pool_size=5,
    max_overflow=5,          # hard ceiling of 10 per process, not 15
    pool_timeout=10,         # fail fast; a queued worker is a stalled worker
    pool_recycle=1800,       # under a proxy or a managed DB, assume idle kills
    pool_pre_ping=True,      # one cheap round trip beats one lost batch
    insertmanyvalues_page_size=500,
)

pool_pre_ping=True is the documented pessimistic disconnect strategy: SQLAlchemy emits a dialect-specific ping (typically SELECT 1) on checkout and, if it fails, discards that connection and invalidates every pooled connection older than the current moment. It costs a round trip per checkout and removes the entire class of "the first query after an idle period fails". pool_recycle is the companion for backends that close idle connections on a timer — the documentation names it as the immediate fix for MySQL's server has gone away, and the same reasoning applies to any managed Postgres behind a connection proxy.

Forking is where pools go wrong

Multiprocessing crawlers hit a specific, documented hazard: pooled connections are not shared with a forked process. Two processes end up writing down the same socket, and the symptoms are corrupted protocol state rather than a clean error. The official remedy is to dispose the inherited pool in the child, without closing the parent's sockets:

from multiprocessing import Pool


def init_worker() -> None:
    # close=False drops the inherited connection references WITHOUT closing
    # them — the parent still owns those sockets. The child then opens its own.
    engine.dispose(close=False)


with Pool(processes=8, initializer=init_worker) as pool:
    pool.map(scrape_one, urls)

One session per unit of work

The Session is documented as "a mutable, stateful object that represents a single database transaction", and it "cannot be shared among concurrent threads or asyncio tasks without careful synchronization". Read that as a design rule rather than a warning: a session's lifetime is a transaction's lifetime, which is how long you hold locks.

Create the sessionmaker once at module scope and open a session per unit of work:

from sqlalchemy.orm import sessionmaker

# Once, at import time. The factory is cheap and shareable; sessions are not.
SessionFactory = sessionmaker(engine)


def store_batch(rows: list[dict[str, object]]) -> None:
    # One transaction per BATCH. Per row, commit overhead dominates the run;
    # per crawl, one bad page rolls back an hour of work while every row you
    # did write stays locked away from readers.
    with SessionFactory() as session, session.begin():
        session.execute(upsert_pages(rows))

session.begin() as a context manager commits on success and rolls back on any exception, so there is no code path where a half-written batch escapes. Batches of roughly 200 to 1,000 rows are the usual sweet spot: large enough that round trips stop dominating, small enough that a rollback is cheap and lock duration stays in the tens of milliseconds.

Make every write idempotent

This is the section that pays for the article. A crawler is a system that will re-run: retries, backfills, an operator restarting yesterday's job. If writing the same page twice produces duplicates or clobbers good data, every one of those events becomes an incident. INSERT ... ON CONFLICT DO UPDATE moves the deduplication into the database, where it is atomic.

from sqlalchemy import func
from sqlalchemy.dialects.postgresql import insert


def upsert_pages(rows: list[dict[str, object]]):
    stmt = insert(ScrapedPage).values(rows)
    return stmt.on_conflict_do_update(
        # Infers the unique index behind the natural key.
        index_elements=[ScrapedPage.source, ScrapedPage.external_id],
        # `excluded` is the row PostgreSQL tried to insert — the fresh scrape.
        set_={
            "url": stmt.excluded.url,
            "title": stmt.excluded.title,
            "payload": stmt.excluded.payload,
            "content_hash": stmt.excluded.content_hash,
            "last_changed_at": func.now(),
        },
        # The line that pays for itself: a re-crawl that found nothing new
        # writes no row at all. No dead tuple, no WAL, no index churn.
        where=ScrapedPage.content_hash != stmt.excluded.content_hash,
    )

Note the import path — the upsert lives on the dialect-specific insert, sqlalchemy.dialects.postgresql.insert, not the generic one. SQLite offers the same on_conflict_do_update() spelling, MySQL and MariaDB offer on_duplicate_key_update(), and there is no portable construct that covers all three. That is a genuine reason to develop against the engine you deploy on.

The where clause deserves a moment of honesty about its trade-off. Skipping the update means last_changed_at records when the content last changed, which is what a change-detection pipeline wants — but it also means you are no longer recording when you last checked. If you need both, keep the guarded update for the record itself and write the cheap "seen at" stamp to a separate, narrow table that no analyst joins against.

The pay-off is a change feed you get for free. Because unchanged rows are skipped, RETURNING hands back exactly the rows that moved:

changed = session.scalars(
    upsert_pages(rows).returning(ScrapedPage),
    # Refreshes any of these objects already living in the session's identity
    # map — unlike a plain INSERT, some of these rows already existed.
    execution_options={"populate_existing": True},
).all()

for page in changed:
    enqueue_downstream(page.id)

That is the difference between "we re-index everything nightly" and "we re-index the 0.4% that changed". On a corpus of a million pages it is the difference between a job that runs in minutes and one that runs all night.

Bulk writes: let insertmanyvalues do the batching

SQLAlchemy 2.0 accepts a list of dictionaries as the parameter set for an insert(), and the ORM interprets the keys as attribute names rather than column names — a deliberate 2.0 change that matters the moment a mapped attribute and its column are spelled differently.

from sqlalchemy import insert

session.execute(
    # render_nulls keeps every row in one batch. Scraped records are ragged by
    # nature — an optional field that is None would otherwise split the run
    # into several statements at exactly the rows you have most of.
    insert(ScrapedPage).execution_options(render_nulls=True),
    rows,
)

Underneath, the insertmanyvalues feature rewrites that into batched multi-row INSERT ... VALUES statements. It is on by default for PostgreSQL, MySQL, SQLite, SQL Server and Oracle, and the batch size follows insertmanyvalues_page_size, which "defaults to 1000, but may also be subject to dialect-specific limiting factors". Tune it down when rows carry large JSON payloads: the win comes from fewer round trips, and a statement too large for the server's parameter limits gives that win straight back.

When you need the generated primary keys, ask for them in parameter order:

page_ids = session.scalars(
    # sort_by_parameter_order guarantees the returned ids line up with the
    # input rows, which is what lets you attach child records without a
    # second SELECT. Added in SQLAlchemy 2.0.10.
    insert(ScrapedPage).returning(ScrapedPage.id, sort_by_parameter_order=True),
    rows,
).all()

Two things not to do. Do not loop session.add() over ten thousand objects and commit once — you pay full unit-of-work bookkeeping for rows nobody will mutate. And do not build the SQL string yourself to "avoid the ORM overhead"; the parameterised path is what keeps scraped text — which is untrusted input by definition — out of your SQL grammar.

Async pipelines: one AsyncSession per task

If your fetcher is already asyncio — most modern ones are, and our notes on proxies for AI agents cover why that shape has spread — running the database on the same event loop avoids a thread hop. The rules are narrow and non-negotiable.

import asyncio
import os

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

# The URL carries the async driver: postgresql+asyncpg://...
engine = create_async_engine(os.environ["SCRAPER_DATABASE_URL"], pool_size=10, max_overflow=0)

# expire_on_commit=False so attributes stay readable after commit. The default
# would expire them, and refreshing an expired attribute is implicit IO —
# exactly what asyncio cannot do behind your back.
AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def store(rows: list[dict[str, object]], gate: asyncio.Semaphore) -> None:
    # One AsyncSession per task: a single instance is explicitly documented as
    # unsafe in concurrent tasks. The semaphore keeps the task count from
    # outrunning the pool, which is the other half of the same budget.
    async with gate, AsyncSessionFactory() as session, session.begin():
        await session.execute(upsert_pages(rows))


async def main(batches: list[list[dict[str, object]]]) -> None:
    gate = asyncio.Semaphore(10)
    try:
        await asyncio.gather(*(store(batch, gate) for batch in batches))
    finally:
        # Closes the pool. Skip it and the loop shuts down under open sockets.
        await engine.dispose()

The error you will meet if you get it wrong is MissingGreenlet, and it almost always means a lazy load fired outside the async context. The asyncio documentation offers three fixes, in the order I would try them: eager-load with selectinload() in the query, add the AsyncAttrs mixin to your base and await obj.awaitable_attrs.things, or drop into session.run_sync() for a block of ordinary synchronous ORM code.

One caution on sizing. Async does not raise your database's connection limit — it makes it far easier to reach. A thousand concurrent tasks against pool_size=10 is not a deadlock, but every task after the tenth is queued, and pool_timeout decides whether that queue fails loudly or silently becomes your latency.

Reading it back without an N+1

The read path is where scraped data gets exported, and it is where the classic ORM performance bug lives: iterating parents and touching a relationship on each one emits one query per parent. SQLAlchemy's answer is to state the loading strategy in the query.

from sqlalchemy import select
from sqlalchemy.orm import raiseload, selectinload

stmt = (
    select(ScrapedPage)
    .options(
        # selectinload is the documented default for COLLECTIONS: a second
        # SELECT with an IN clause, so no join multiplies the parent rows.
        selectinload(ScrapedPage.observations),
        # Everything not eager-loaded above now RAISES instead of quietly
        # emitting a query. An N+1 becomes a test failure, not a slow export.
        raiseload("*"),
    )
    .where(ScrapedPage.source == "example-retailer")
    .order_by(ScrapedPage.last_changed_at.desc())
    .limit(500)
)
pages = session.scalars(stmt).all()

For a scalar many-to-one — an observation back to its page — joinedload() is the better shape, because the join adds one column set rather than a second round trip. The one trap the documentation is explicit about: joinedload() on a collection multiplies rows, so results must be de-duplicated with .unique() before you count anything.

raiseload("*") is the underrated one. Turn it on in tests and in any batch export and every accidental lazy load becomes a loud, located failure during development instead of a mysterious hour-long job in production.

Observability: measure the SQL, not the vibe

echo=True is a development switch. In production it logs every statement — including bound parameters, which in a scraper means scraped content and, on the connection itself, credentials. Use events instead, and log the statement without the parameters.

import logging
import time

from sqlalchemy import event

log = logging.getLogger("scraper.sql")
SLOW_QUERY_SECONDS = 0.5


@event.listens_for(engine, "before_cursor_execute")
def _start_timer(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault("query_start", []).append(time.perf_counter())


@event.listens_for(engine, "after_cursor_execute")
def _record_duration(conn, cursor, statement, parameters, context, executemany):
    elapsed = time.perf_counter() - conn.info["query_start"].pop()
    if elapsed >= SLOW_QUERY_SECONDS:
        # The statement, never `parameters`: that tuple is the payload.
        log.warning("slow sql %.3fs executemany=%s %s", elapsed, executemany, statement)

The event signature is fixed — (conn, cursor, statement, parameters, context, executemany) for both hooks — and the stack on conn.info is what keeps nested executions paired correctly. Three metrics are worth exporting from here: statement duration by operation, executemany share (your batching is working when it rises), and pool checkout wait time, which is the number that tells you whether the crawler is slow because of the target site or because of your own pool_size.

Retry the transaction, never the statement

The documentation is blunt about this: "when a connection is lost, the entire transaction is lost. There is no useful way that the database can reconnect and retry and continue where it left off." The recommended shape is to "retry the entire operation from the start of the transaction".

import random
import time
from collections.abc import Callable
from typing import TypeVar

from sqlalchemy.exc import DBAPIError

T = TypeVar("T")
# Serialization failure and deadlock detected: transient by definition.
RETRYABLE_SQLSTATES = frozenset({"40001", "40P01"})


def _is_retryable(error: DBAPIError) -> bool:
    if error.connection_invalidated:
        return True
    orig = error.orig
    return getattr(orig, "sqlstate", None) in RETRYABLE_SQLSTATES


def with_retry(work: Callable[[], T], attempts: int = 5) -> T:
    """Re-run a whole unit of work, from BEGIN, on a transient failure."""
    for attempt in range(1, attempts + 1):
        try:
            return work()
        except DBAPIError as error:
            if attempt == attempts or not _is_retryable(error):
                raise
            # Full jitter. A fleet that backs off on one schedule reconnects
            # as a thundering herd and knocks the database over a second time.
            time.sleep(random.uniform(0, min(2**attempt * 0.1, 5.0)))
    raise AssertionError("unreachable")


with_retry(lambda: store_batch(rows))

This is safe only because the write is an upsert. Retrying a plain INSERT after an ambiguous failure is how you get duplicates; retrying an ON CONFLICT DO UPDATE converges on the same row no matter how many times it runs. Idempotency is not a nicety here — it is the precondition that makes automatic recovery possible at all.

pool_pre_ping and this decorator solve different halves of the problem. Pre-ping catches the connection that died while idle in the pool, before you have started work. The retry catches the connection that dies mid-transaction, when there is nothing to salvage.

Schema changes: autogenerate is a draft, not a migration

Alembic is the migration tool from the same project, and its own documentation says autogenerate "is not intended to be perfect" and that "it is always necessary to manually review and correct the candidate migrations that autogenerate produces". Take that literally.

alembic revision --autogenerate -m "add price_observation"
alembic upgrade head

It reliably detects table and column additions and removals, changes of nullability, basic index and named unique-constraint changes, basic foreign-key changes, and named check constraints. It does not detect table renames or column renames — both surface as a drop plus an add, which on a scraped corpus means deleting the column and rebuilding it empty — and it cannot see anonymously named constraints at all. That is why the naming_convention sits on the MetaData in the first code block: without it, autogenerate has nothing to match constraints by.

The workflow that holds up: generate, read the file line by line, rewrite drop-plus-add pairs into op.alter_column(..., new_column_name=...), and run the migration against a restored copy of production before it goes anywhere near production.

Testing: roll back instead of cleaning up

Scraper tests need a database that behaves like the real one and a fixture that does not leave residue. SQLAlchemy 2.0 documents the pattern: open an outer transaction on a connection, bind the session to that connection in savepoint mode, and roll the outer transaction back when the test ends.

import pytest
from sqlalchemy.orm import Session


@pytest.fixture()
def session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    # join_transaction_mode="create_savepoint" lets the CODE UNDER TEST call
    # session.commit() for real — the commit lands on a SAVEPOINT inside the
    # outer transaction, which the fixture then throws away wholesale.
    db = Session(bind=connection, join_transaction_mode="create_savepoint")
    try:
        yield db
    finally:
        db.close()
        transaction.rollback()
        connection.close()

Run it against the same engine you deploy on. SQLite is a fine target for the pure-Python parts, but ON CONFLICT semantics, JSONB operators and NULL ordering all differ, and a test suite that passes on SQLite while production runs Postgres is testing a different program.

Five failures I keep finding in scraper databases

None of these are exotic. They are the same five every time, and four of the five are configuration rather than code.

What this buys the fetching side

A storage layer built this way changes what the crawler is allowed to do. Because writes are idempotent, a worker can crash and be restarted without reconciliation. Because the upsert reports what changed, the scheduler can crawl frequently-changing pages more often and quiet ones less — which is the single biggest lever on proxy spend, and it pairs directly with the politeness practices in how to scrape without getting blocked. And because the change feed exists at all, downstream jobs like price monitoring consume a stream instead of re-reading a table.

The fetching stage gets the attention because it is where the blocks and the bills are. The storage stage is where you decide how much fetching you have to pay for.