Saving Scraped Data to PostgreSQL

Once a crawl produces more rows than you want to re-derive, a real database beats a pile of files, and this page — part of Storing and Exporting Scraped Data — covers loading scraped records into PostgreSQL efficiently and idempotently with psycopg 3.

The pattern that works is short: give the table a unique key derived from the source, batch rows in the hundreds, write them with a single multi-row statement carrying ON CONFLICT ... DO UPDATE, and commit once per batch. Batching turns thousands of network round-trips into a handful, the upsert makes the loader safe to re-run after any failure, and one commit per batch keeps each write atomic. Above roughly a hundred thousand rows per run, COPY into a staging table followed by one INSERT ... SELECT is faster still.

Why Batching and Upserts Decide the Throughput

Inserting one row per statement is the classic scraper bottleneck, and the reason is latency rather than database speed. Every INSERT is a request and a response. Against a database on the same host that round-trip might be 0.2 ms; against a managed instance in the same region it is closer to 1–2 ms. Ten thousand single-row inserts therefore cost 10–20 seconds of pure waiting before PostgreSQL does any work at all — often longer than the crawl that produced the rows. Batch 500 rows into one statement and the same load is twenty round-trips.

Committing per row multiplies the problem, because each commit forces a write-ahead log flush to disk. One commit per 500-row batch is dramatically cheaper and has the useful property that a failure mid-batch rolls the whole batch back rather than leaving a half-written page of results.

Idempotency is the second requirement, and it is not optional for a scraper. Crawls are retried, workers are restarted, and schedules overlap, so the same record will arrive more than once. Without a unique constraint you silently accumulate duplicates; with one and no conflict handling, a single collision aborts the entire batch with UniqueViolation. ON CONFLICT resolves both cases. This is what makes the storage layer safe to sit behind the retry logic in Retrying Failed Requests with Tenacity and behind at-least-once task queues.

The Conflict Target

ON CONFLICT (source_url) is not a filter — it names a unique index, and PostgreSQL uses that index to detect the collision. If no unique index or constraint covers exactly those columns, the statement fails outright rather than falling back to a plain insert.

Choose the key carefully, because it defines what "the same record" means for the rest of the project's life. A canonicalised source URL is the usual choice for page-per-record scrapes; a site's own product or listing identifier is better when one record is reachable from several URLs. Whatever you pick, normalise it before it reaches the database — strip tracking parameters, lowercase the host, and settle on whether the trailing slash is present — or ?utm_source= will manufacture duplicates that no constraint can catch. Normalisation of the values themselves is covered in Normalizing Prices, Dates and Units.

Bulk Upsert with psycopg 3

In psycopg 3 the helper to reach for is cursor.executemany, which pipelines the whole batch to the server in one exchange rather than issuing one round-trip per row. (The execute_values helper that older tutorials use lives in psycopg2.extras and does not exist in psycopg 3.)

import datetime as dt

import httpx
import psycopg
from selectolax.parser import HTMLParser

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/125.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml",
}
DSN = "postgresql://scraper:secret@localhost:5432/scrapes"

DDL = """
CREATE TABLE IF NOT EXISTS products (
    id          BIGSERIAL PRIMARY KEY,
    source_url  TEXT NOT NULL,
    title       TEXT NOT NULL,
    price_cents INTEGER,
    scraped_at  TIMESTAMPTZ NOT NULL,
    CONSTRAINT products_source_url_key UNIQUE (source_url)
);
"""

UPSERT = """
INSERT INTO products (source_url, title, price_cents, scraped_at)
VALUES (%s, %s, %s, %s)
ON CONFLICT (source_url) DO UPDATE SET
    title       = EXCLUDED.title,
    price_cents = EXCLUDED.price_cents,
    scraped_at  = EXCLUDED.scraped_at
WHERE products.title IS DISTINCT FROM EXCLUDED.title
   OR products.price_cents IS DISTINCT FROM EXCLUDED.price_cents;
"""

Row = tuple[str, str, int | None, dt.datetime]


def scrape(url: str) -> list[Row]:
    now = dt.datetime.now(dt.timezone.utc)
    with httpx.Client(headers=HEADERS, timeout=15.0, follow_redirects=True) as client:
        resp = client.get(url)
    resp.raise_for_status()
    tree = HTMLParser(resp.text)
    rows: list[Row] = []
    for card in tree.css("article.product_pod"):
        link = card.css_first("h3 a")
        price = card.css_first("p.price_color")
        if link is None or price is None:
            continue
        cents = int(round(float(price.text(strip=True).lstrip("£")) * 100))
        rows.append((url + link.attributes.get("href", ""), link.attributes.get("title", ""), cents, now))
    return rows


def save(rows: list[Row]) -> int:
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            cur.execute(DDL)
            cur.executemany(UPSERT, rows)
            written = cur.rowcount
        conn.commit()
    return written


if __name__ == "__main__":
    data = scrape("https://books.toscrape.com/")
    print(f"upserted {save(data)} of {len(data)} rows")

Three details are load-bearing. EXCLUDED is the pseudo-table holding the row that would have been inserted, so the DO UPDATE branch copies the fresh values over the stored ones. The WHERE ... IS DISTINCT FROM clause suppresses no-op updates: without it, every re-scrape rewrites every row, generating dead tuples that autovacuum then has to clean up, and bumping scraped_at on records that did not actually change. And every value is passed as a parameter, never formatted into the SQL string — scraped text contains quotes, backslashes and occasionally deliberate injection attempts.

COPY When the Batch Is Large

Above roughly a hundred thousand rows a run, the fastest correct path is PostgreSQL's binary COPY into an unlogged staging table, then a single set-based upsert out of it. COPY skips per-row statement parsing entirely.

import datetime as dt

import psycopg

DSN = "postgresql://scraper:secret@localhost:5432/scrapes"

STAGE = """
CREATE UNLOGGED TABLE IF NOT EXISTS products_stage (
    source_url  TEXT,
    title       TEXT,
    price_cents INTEGER,
    scraped_at  TIMESTAMPTZ
);
TRUNCATE products_stage;
"""

MERGE = """
INSERT INTO products (source_url, title, price_cents, scraped_at)
SELECT DISTINCT ON (source_url) source_url, title, price_cents, scraped_at
FROM products_stage
ORDER BY source_url, scraped_at DESC
ON CONFLICT (source_url) DO UPDATE SET
    title       = EXCLUDED.title,
    price_cents = EXCLUDED.price_cents,
    scraped_at  = EXCLUDED.scraped_at;
"""


def bulk_load(rows: list[tuple[str, str, int | None, dt.datetime]]) -> None:
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            cur.execute(STAGE)
            copy_sql = "COPY products_stage (source_url, title, price_cents, scraped_at) FROM STDIN"
            with cur.copy(copy_sql) as copy:
                for row in rows:
                    copy.write_row(row)
            cur.execute(MERGE)
        conn.commit()


if __name__ == "__main__":
    now = dt.datetime.now(dt.timezone.utc)
    bulk_load([("https://example.com/p/1", "Widget", 999, now)])
    print("bulk load complete")

The DISTINCT ON is not decoration. A single INSERT ... ON CONFLICT statement cannot update the same target row twice — if the staging table contains two rows with the same source_url, PostgreSQL raises ON CONFLICT DO UPDATE command cannot affect row a second time. Deduplicating in the SELECT, keeping the most recent scrape, is the fix. The same error appears in the executemany path only if you have collapsed the batch into one statement yourself.

The Indexes Worth Paying For

Every index is maintained on every write, so a scrape table should carry the ones it earns and no more.

The unique index on the conflict target is mandatory — the upsert does not work without it. A descending index on scraped_at is the one most crawls want next, because "what changed since the last run" is the query you will actually write, and it is what makes incremental strategies like those in Incremental Crawls with ETag and Last-Modified cheap to drive from the database side. A GIN index on a JSONB payload column is powerful and expensive; add it only when you have a query that needs it, and prefer jsonb_path_ops, which is smaller and faster than the default operator class for containment queries.

If you are loading tens of millions of rows into a fresh table, drop the non-unique indexes first and create them after the load. Building an index once over a full table is much faster than maintaining it row by row.

Connection Pooling for Concurrent Workers

PostgreSQL forks a backend process per connection, so connections are expensive and hard-capped by max_connections (often 100 on a default install, and lower on small managed instances). A crawl with concurrent workers must not open one per task.

import datetime as dt

from psycopg_pool import ConnectionPool

DSN = "postgresql://scraper:secret@localhost:5432/scrapes"

UPSERT = """
INSERT INTO products (source_url, title, price_cents, scraped_at)
VALUES (%s, %s, %s, %s)
ON CONFLICT (source_url) DO UPDATE SET
    title       = EXCLUDED.title,
    price_cents = EXCLUDED.price_cents,
    scraped_at  = EXCLUDED.scraped_at;
"""

pool = ConnectionPool(DSN, min_size=2, max_size=10, open=False)


def save_batch(rows: list[tuple[str, str, int | None, dt.datetime]]) -> None:
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.executemany(UPSERT, rows)
        conn.commit()


if __name__ == "__main__":
    pool.open()
    now = dt.datetime.now(dt.timezone.utc)
    save_batch([("https://example.com/p/1", "Widget", 999, now)])
    print("saved via pool")
    pool.close()

Size max_size against the server, not against your worker count: the sum of max_size across every process that connects must stay below max_connections, with headroom for your own psql sessions and for monitoring. Twenty containers each holding a pool of ten is two hundred connections, which will produce FATAL: sorry, too many clients already on a default configuration. Past a few dozen total connections, put PgBouncer in transaction mode in front of the database and keep the application pools small — but note that transaction pooling forbids server-side prepared statements, so set prepare_threshold=None on the connection when you do.

The SQLAlchemy Alternative

If you want typed models, Alembic migrations, or code that also has to run against SQLite in tests, SQLAlchemy exposes the same upsert through the PostgreSQL dialect. You give up some raw throughput for schema management and easier querying later.

import datetime as dt

from sqlalchemy import DateTime, Integer, String, create_engine
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

engine = create_engine("postgresql+psycopg://scraper:secret@localhost:5432/scrapes")


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products_orm"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    source_url: Mapped[str] = mapped_column(String, unique=True)
    title: Mapped[str] = mapped_column(String)
    scraped_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True))


def upsert(rows: list[dict[str, object]]) -> None:
    Base.metadata.create_all(engine)
    stmt = insert(Product).values(rows)
    stmt = stmt.on_conflict_do_update(
        index_elements=[Product.source_url],
        set_={"title": stmt.excluded.title, "scraped_at": stmt.excluded.scraped_at},
    )
    with engine.begin() as conn:
        conn.execute(stmt)


if __name__ == "__main__":
    now = dt.datetime.now(dt.timezone.utc)
    upsert([{"source_url": "https://example.com/p/9", "title": "Gadget", "scraped_at": now}])
    print("orm upsert done")

Whichever driver you use, validate records before they reach the database rather than relying on column constraints for error messages — a typed model at the boundary, as in Validating Scraped Data with Pydantic, turns a database exception into a clear per-field error you can log and skip. In a Scrapy project this belongs in an item pipeline; see Writing Scrapy Item Pipelines.

Edge Cases and Caveats

• ON CONFLICT needs a real unique index. A plain NOT NULL column is not enough. Without a matching constraint the statement raises there is no unique or exclusion constraint matching the ON CONFLICT specification.
• You cannot update the same row twice in one statement. Duplicate keys within a single batch raise ON CONFLICT DO UPDATE command cannot affect row a second time. Deduplicate in Python or with DISTINCT ON before the write.
• Batch size has a sweet spot. Somewhere between 500 and 2,000 rows is usually right. Too small and you lose the batching win; too large and you build a multi-megabyte statement, hold a long transaction and inflate the memory of both client and server.
• Store timezone-aware timestamps. Use TIMESTAMPTZ and datetime.now(dt.timezone.utc). Naive timestamps from workers in different regions sort incorrectly and are impossible to fix retrospectively.
• Money is not a float. Store minor units in an INTEGER or use NUMERIC. float prices accumulate representation error and compare unequal to themselves after a round trip.
• Long transactions block vacuum. Holding one transaction open across a whole crawl prevents autovacuum from cleaning dead tuples site-wide, and bloats the table. Commit per batch and let the connection go back to the pool.
• Nulls are not empty strings. A missing field should be NULL, not "", or you cannot distinguish "not present on the page" from "present and blank" — a difference that matters when you later diff runs.

Frequently Asked Questions

Why batch inserts instead of a loop of single INSERT statements? Because the cost is dominated by network round-trips, not by PostgreSQL. Each single-row insert is one request and one response, so ten thousand rows against a database 1 ms away is at least ten seconds of pure latency. One statement carrying five hundred rows collapses that to twenty exchanges, and one commit per batch avoids ten thousand write-ahead log flushes.

How does ON CONFLICT make my loader idempotent? It tells PostgreSQL what to do when an inserted row collides with an existing unique key: DO UPDATE overwrites the stored row with the new values, and DO NOTHING keeps the original. Either way the statement succeeds instead of aborting, so re-running a scraper refreshes records rather than erroring or duplicating, and overlapping or retried crawls become safe.

Do I need a connection pool for a single-threaded scraper? No. One long-lived connection is fine and simpler. A pool becomes necessary once concurrent tasks write at the same time, because opening a connection per task is slow — PostgreSQL forks a backend process for each — and enough of them will exhaust the server's connection limit outright.

Should I use psycopg directly or SQLAlchemy? Use psycopg when you want maximum bulk-write throughput and full control of the SQL, which is the common case for a loader. Choose SQLAlchemy when you value typed models, Alembic migrations and database portability, and can absorb a modest overhead. Both support PostgreSQL upserts, so idempotency is available either way.

Related

• Storing and Exporting Scraped Data — the parent topic on choosing a sink for a crawl's output.
• Exporting Scraped Data to CSV and Parquet — file formats for handing data on once it is in the database.
• Validating Scraped Data with Pydantic — catching bad records before they reach a column constraint.
• Deduplicating Records with Fuzzy Matching — for duplicates a unique key cannot see.