SQLAlchemy für Web Scraping: eine produktionsreife Datenschicht

Kurz gesagt: In einer Scraping-Pipeline ist SQLAlchemy nicht der Teil, der abruft — es ist der Teil, der ein erneutes Abrufen überflüssig macht. Modellieren Sie den Datensatz, den Sie wirklich brauchen, halten Sie den Connection-Pool unter dem Limit der Datenbank, schreiben Sie jede Zeile über ein ON CONFLICT DO UPDATE, damit ein Retry nichts kostet, und wiederholen Sie die gesamte Transaktion statt eines einzelnen Statements. Alles Folgende bezieht sich auf SQLAlchemy 2.0 — 2.0.52, veröffentlicht am 11. August 2026 — und nennt die offiziellen Standardwerte, statt sie zu raten.

Wo die Datenbank in einer Scraping-Pipeline sitzt

Eine produktive Pipeline hat fünf Stufen — URL-Beschaffung, Abruf, Parsing, Validierung und Speicherung —, und unser Web-Scraping-Leitfaden geht die ganze Kette durch. Nur eine dieser Stufen kostet Geld pro Byte. Residential-Proxy-Traffic wird pro Gigabyte abgerechnet, also entscheidet sich die Wirtschaftlichkeit der Pipeline daran, wie oft Sie eine Seite ein zweites Mal abrufen müssen; unsere Kostenschätzung für Web Scraping hinterlegt das mit echten Zahlen.

Das ordnet die Datenschicht neu ein. Ihre Aufgabe ist nicht, die Zeilen irgendwo abzulegen. Ihre Aufgabe ist, drei Fragen so billig zu beantworten, dass der Abrufteil es nie tun muss:

1. Habe ich diesen Datensatz schon gesehen, und hat er sich geändert? Lautet die Antwort nein, spart der Crawl eine kostenpflichtige Anfrage.
2. Was überlebt einen Absturz? Ein Worker, der mitten im Batch stirbt, darf nicht die halbe Seite committen und die andere Hälfte verlieren.
3. Können Analysten lesen, während der Crawler schreibt? Eine lange Transaktion, die Sperren hält, macht Warehouse und Crawler zu Gegnern.

SQLAlchemy beantwortet alle drei, in zwei Schichten, die man auseinanderhalten sollte. Core ist die SQL-Ausdruckssprache: Table, select(), insert(), die Engine und der Connection-Pool. Das ORM legt darauf gemappte Klassen, eine Identity Map und die Unit-of-Work-Session. Die produktive Aufteilung für einen Scraper: das Schema mit deklarativen ORM-Modellen definieren — dann haben Python-Typen und DDL eine gemeinsame Quelle — und Batches mit Core-nahen Statements schreiben, denn zehntausend Zeilen einzufügen ist eine Mengenoperation, keine zehntausend Objektmutationen.

Modellieren Sie den Datensatz, nicht die gescrapte Seite

Der teuerste Modellierungsfehler in einem Scraper ist, das Artefakt statt der Tatsache zu speichern. Rohes HTML ist riesig, ändert sich aus Gründen, die Sie nicht interessieren, und lässt sich nicht abfragen. Speichern Sie den extrahierten Datensatz plus genug Metadaten, um zu entscheiden, ob er neu ist.

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")

Drei Details darin leisten die meiste Arbeit.

Mapped[] entscheidet über die Nullbarkeit. Die dokumentierte Regel lautet: Mapped[str] erzeugt NOT NULL, Mapped[Optional[str]] — oder Mapped[str | None] — erzeugt NULL, und primary_key=True impliziert unabhängig davon NOT NULL. Sie können das mit mapped_column(nullable=...) überschreiben, aber der Standard bedeutet, dass Ihr Typprüfer dieselbe Wahrheit liest wie Ihr Schema.

type_annotation_map zentralisiert die Python-zu-SQL-Zuordnung. SQLAlchemy mappt bereits int auf Integer, str auf String, datetime.datetime auf DateTime, uuid.UUID auf Uuid und so weiter. Diese Map auf der Basisklasse zu überschreiben ist der Weg, eine projektweite Richtlinie durchzusetzen — zeitzonenbewusste Zeitstempel, JSONB statt generischem JSON — ohne sie auf 40 Spalten zu wiederholen.

Der natürliche Schlüssel ist ein echtes Constraint, keine Konvention. source plus external_id macht dieselbe Produktseite über Crawls hinweg wiedererkennbar. Eine reine Surrogat-id kann das nicht ausdrücken, und alles im nächsten Abschnitt hängt daran, dass die Datenbank die Kollision selbst erkennen kann.

Engine und Pool sind Ihr Nebenläufigkeitsbudget

create_engine() ist eine Factory, keine Verbindung: Sie hält einen Connection-Pool, der verzögert erzeugt und geteilt wird. Außer bei SQLite in-memory ist der Standardpool QueuePool, und seine dokumentierten Standardwerte sind die Zahlen, die Sie zuerst treffen.

Die entscheidende Rechnung ist Worker × (pool_size + max_overflow) ≤ Verbindungslimit − Reserve. Mit den Standardwerten fordern acht Crawler-Prozesse 120 Verbindungen von einer Datenbank, die vielleicht 100 erlaubt, und der Fehler zeigt sich als ablaufendes pool_timeout in den Workern, während eine Migration überhaupt keine Verbindung mehr bekommt. Setzen Sie die Obergrenze bewusst:

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 ist die dokumentierte pessimistische Strategie gegen Verbindungsabbrüche: SQLAlchemy sendet beim Checkout einen dialektspezifischen Ping (typischerweise SELECT 1) und verwirft die Verbindung bei Fehlschlag, samt Invalidierung aller älteren gepoolten Verbindungen. Das kostet eine Rundreise pro Checkout und beseitigt die gesamte Fehlerklasse „die erste Abfrage nach einer Ruhephase schlägt fehl". pool_recycle ist das Gegenstück für Backends, die inaktive Verbindungen nach Zeit schließen — die Dokumentation nennt es als Sofortmaßnahme gegen MySQLs server has gone away, und dieselbe Logik gilt für jedes verwaltete PostgreSQL hinter einem Verbindungsproxy.

Beim Forken gehen Pools kaputt

Multiprocessing-Crawler treffen eine spezifische, dokumentierte Gefahr: gepoolte Verbindungen werden nicht an einen geforkten Prozess weitergegeben. Zwei Prozesse schreiben dann in denselben Socket, und das Symptom ist beschädigter Protokollzustand statt eines sauberen Fehlers. Das offizielle Gegenmittel ist, den geerbten Pool im Kind zu verwerfen, ohne die Sockets des Elternprozesses zu schließen:

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)

Eine Session pro Arbeitseinheit

Die Session ist dokumentiert als „a mutable, stateful object that represents a single database transaction", und sie „cannot be shared among concurrent threads or asyncio tasks without careful synchronization". Lesen Sie das als Entwurfsregel, nicht als Warnhinweis: Die Lebensdauer einer Session ist die einer Transaktion, und die ist die Zeit, in der Sperren gehalten werden.

Erzeugen Sie den sessionmaker einmal auf Modulebene und öffnen Sie eine Session pro Arbeitseinheit:

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() als Kontextmanager committet bei Erfolg und rollt bei jeder Ausnahme zurück, es gibt also keinen Pfad, auf dem ein halb geschriebener Batch entkommt. Batches von etwa 200 bis 1.000 Zeilen sind der übliche Sweet Spot: groß genug, dass Rundreisen nicht mehr dominieren, klein genug, dass ein Rollback billig bleibt und die Sperrdauer im Bereich von Millisekunden liegt.

Machen Sie jeden Schreibvorgang idempotent

Dieser Abschnitt rechtfertigt den Artikel. Ein Crawler ist ein System, das erneut läuft: Retries, Backfills, ein Operator, der den gestrigen Job neu startet. Wenn dasselbe Schreiben zweimal Duplikate erzeugt oder gute Daten zerstört, wird jedes dieser Ereignisse zum Incident. INSERT ... ON CONFLICT DO UPDATE verschiebt die Deduplizierung in die Datenbank, wo sie atomar ist.

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,
    )

Beachten Sie den Importpfad — das Upsert liegt am dialektspezifischen insert, also sqlalchemy.dialects.postgresql.insert, nicht am generischen. SQLite bietet dieselbe Schreibweise on_conflict_do_update(), MySQL und MariaDB bieten on_duplicate_key_update(), und es gibt kein portables Konstrukt, das alle drei abdeckt. Das ist ein handfester Grund, gegen die Engine zu entwickeln, die Sie ausrollen.

Zur where-Bedingung gehört ein ehrliches Wort über den Kompromiss. Das Update zu überspringen bedeutet, dass last_changed_at festhält, wann sich der Inhalt zuletzt geändert hat — genau das will eine Änderungserkennung —, aber Sie halten damit nicht mehr fest, wann Sie zuletzt nachgesehen haben. Brauchen Sie beides, behalten Sie das abgesicherte Update für den Datensatz und schreiben Sie den billigen Zeitstempel „zuletzt gesehen" in eine separate, schmale Tabelle, gegen die niemand joint.

Die Belohnung ist ein Änderungs-Feed, den Sie geschenkt bekommen. Weil unveränderte Zeilen übersprungen werden, liefert RETURNING exakt die Zeilen, die sich bewegt haben:

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)

Das ist der Unterschied zwischen „wir indexieren nachts alles neu" und „wir indexieren die 0,4 % neu, die sich geändert haben". Bei einer Million Seiten ist das der Unterschied zwischen einem Job von wenigen Minuten und einem, der die ganze Nacht läuft.

Bulk-Writes: lassen Sie insertmanyvalues bündeln

SQLAlchemy 2.0 akzeptiert eine Liste von Dictionaries als Parametersatz für ein insert(), und das ORM interpretiert die Schlüssel als Attributnamen statt als Spaltennamen — eine bewusste Änderung in 2.0, die in dem Moment zählt, in dem ein gemapptes Attribut anders heißt als seine Spalte.

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,
)

Darunter schreibt die Funktion insertmanyvalues das in gebündelte mehrzeilige INSERT ... VALUES-Statements um. Sie ist standardmäßig aktiv für PostgreSQL, MySQL, SQLite, SQL Server und Oracle, und die Batchgröße folgt insertmanyvalues_page_size, das „defaults to 1000, but may also be subject to dialect-specific limiting factors". Senken Sie den Wert, wenn Zeilen große JSON-Payloads tragen: Der Gewinn kommt aus weniger Rundreisen, und ein Statement jenseits der Parametergrenzen des Servers gibt genau diesen Gewinn zurück.

Wenn Sie die erzeugten Primärschlüssel brauchen, fordern Sie sie in Parameterreihenfolge an:

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()

Zwei Dinge lassen Sie besser: Schleifen Sie nicht session.add() über zehntausend Objekte und committen einmal — Sie bezahlen die volle Unit-of-Work-Buchführung für Zeilen, die niemand mutieren wird. Und bauen Sie das SQL nicht selbst als String zusammen, um „den ORM-Overhead zu sparen"; der parametrisierte Pfad ist genau das, was gescrapten Text — per Definition nicht vertrauenswürdige Eingabe — aus Ihrer SQL-Grammatik heraushält.

Async-Pipelines: eine AsyncSession pro Task

Wenn Ihr Fetcher ohnehin asyncio ist — die meisten modernen sind es, und unsere Notizen zu Proxys für KI-Agenten erklären, warum sich diese Form durchgesetzt hat — vermeidet die Datenbank auf derselben Event-Loop einen Thread-Wechsel. Die Regeln sind eng und nicht verhandelbar.

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()

Der Fehler, den Sie sonst treffen, heißt MissingGreenlet, und fast immer bedeutet er ein Lazy Loading außerhalb des Async-Kontexts. Die asyncio-Dokumentation nennt drei Auswege, in der Reihenfolge, in der ich sie probieren würde: in der Abfrage mit selectinload() eager laden, das AsyncAttrs-Mixin an die Basisklasse hängen und await obj.awaitable_attrs.things schreiben, oder für einen Block gewöhnlichen synchronen ORM-Codes in session.run_sync() wechseln.

Eine Warnung zur Dimensionierung: Async hebt das Verbindungslimit Ihrer Datenbank nicht an — es macht es nur viel leichter, es zu erreichen. Tausend gleichzeitige Tasks gegen pool_size=10 sind kein Deadlock, aber jede Task nach der zehnten wartet in der Schlange, und pool_timeout entscheidet, ob diese Schlange laut scheitert oder still zu Ihrer Latenz wird.

Zurücklesen ohne N+1

Der Lesepfad ist die Stelle, an der gescrapte Daten exportiert werden, und die Stelle, an der der klassische ORM-Performancefehler lebt: über Eltern iterieren und bei jedem eine Beziehung anfassen erzeugt eine Abfrage pro Elternobjekt. SQLAlchemys Antwort ist, die Ladestrategie in der Abfrage zu benennen.

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()

Für eine skalare Many-to-one-Beziehung — eine Beobachtung zurück auf ihre Seite — ist joinedload() die bessere Form, weil der Join einen Spaltensatz hinzufügt statt einer zweiten Rundreise. Die eine Falle, die die Dokumentation ausdrücklich nennt: joinedload() auf einer Collection vervielfacht Zeilen, Ergebnisse müssen also mit .unique() dedupliziert werden, bevor Sie irgendetwas zählen.

raiseload("*") wird unterschätzt. Schalten Sie es in Tests und in jedem Batch-Export ein, und jedes versehentliche Lazy Loading wird zu einem lauten, lokalisierten Fehler während der Entwicklung statt zu einem rätselhaften stundenlangen Job in Produktion.

Observability: messen Sie das SQL, nicht das Gefühl

echo=True ist ein Entwicklungsschalter. In Produktion protokolliert es jedes Statement — inklusive gebundener Parameter, was in einem Scraper gescrapte Inhalte bedeutet und an der Verbindung selbst Zugangsdaten. Nutzen Sie stattdessen Events und protokollieren Sie das Statement ohne die Parameter.

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)

Die Event-Signatur ist fest — (conn, cursor, statement, parameters, context, executemany) für beide Hooks —, und der Stack auf conn.info hält verschachtelte Ausführungen korrekt paarweise. Drei Kennzahlen lohnen den Export: Statement-Dauer nach Operation, der executemany-Anteil (steigt er, funktioniert Ihr Batching), und die Wartezeit beim Pool-Checkout — die Zahl, die Ihnen sagt, ob der Crawler wegen der Zielseite langsam ist oder wegen Ihrer eigenen pool_size.

Wiederholen Sie die Transaktion, nie das Statement

Die Dokumentation ist deutlich: „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." Die empfohlene Form ist, „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))

Das ist nur sicher, weil der Schreibvorgang ein Upsert ist. Ein einfaches INSERT nach einem mehrdeutigen Fehler zu wiederholen erzeugt Duplikate; ein ON CONFLICT DO UPDATE konvergiert auf dieselbe Zeile, egal wie oft es läuft. Idempotenz ist hier keine Annehmlichkeit — sie ist die Voraussetzung dafür, dass automatische Wiederherstellung überhaupt möglich ist.

pool_pre_ping und dieser Dekorator lösen verschiedene Hälften desselben Problems. Pre-Ping fängt die Verbindung ab, die im Pool im Leerlauf gestorben ist, bevor Sie mit der Arbeit begonnen haben. Der Retry fängt die Verbindung ab, die mitten in der Transaktion stirbt, wenn es nichts mehr zu retten gibt.

Schemaänderungen: Autogenerate ist ein Entwurf, keine Migration

Alembic ist das Migrationswerkzeug desselben Projekts, und seine eigene Dokumentation sagt, Autogenerate sei „not intended to be perfect" und „it is always necessary to manually review and correct the candidate migrations that autogenerate produces". Nehmen Sie das wörtlich.

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

Zuverlässig erkannt werden Hinzufügen und Entfernen von Tabellen und Spalten, Änderungen der Nullbarkeit, einfache Änderungen an Indizes und benannten Unique-Constraints, einfache Änderungen an Fremdschlüsseln sowie benannte Check-Constraints. Nicht erkannt werden Tabellen- und Spaltenumbenennungen — beide erscheinen als Drop plus Add, was bei einem gescrapten Korpus heißt, die Spalte zu löschen und leer neu aufzubauen — und anonym benannte Constraints sieht es gar nicht. Deshalb sitzt die naming_convention im ersten Codeblock auf der MetaData: ohne sie hat Autogenerate nichts, woran es Constraints wiedererkennt.

Der Ablauf, der hält: generieren, die Datei Zeile für Zeile lesen, Drop-plus-Add-Paare in op.alter_column(..., new_column_name=...) umschreiben und die Migration gegen eine wiederhergestellte Kopie der Produktion laufen lassen, bevor sie in deren Nähe kommt.

Testen: zurückrollen statt aufräumen

Scraper-Tests brauchen eine Datenbank, die sich wie die echte verhält, und eine Fixture, die keine Rückstände lässt. SQLAlchemy 2.0 dokumentiert das Muster: eine äußere Transaktion auf einer Verbindung öffnen, die Session im Savepoint-Modus an diese Verbindung binden und die äußere Transaktion am Testende zurückrollen.

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()

Lassen Sie das gegen dieselbe Engine laufen, die Sie ausrollen. SQLite ist ein gutes Ziel für die reinen Python-Teile, aber ON CONFLICT-Semantik, JSONB-Operatoren und NULL-Sortierung unterscheiden sich alle, und eine Suite, die auf SQLite grün ist, während Produktion auf Postgres läuft, testet ein anderes Programm.

Fünf Fehler, die ich in Scraper-Datenbanken immer wieder finde

Nichts davon ist exotisch. Es sind jedes Mal dieselben fünf, und vier von fünf sind Konfiguration statt Code.

Was das der Abrufseite bringt

Eine so gebaute Datenschicht ändert, was der Crawler darf. Weil Schreibvorgänge idempotent sind, kann ein Worker abstürzen und ohne Abgleich neu starten. Weil das Upsert meldet, was sich geändert hat, kann der Scheduler häufig wechselnde Seiten öfter und ruhige seltener crawlen — der größte Hebel auf die Proxy-Ausgaben, und er greift direkt in die Praktiken aus Scraping ohne blockiert zu werden. Und weil es den Änderungs-Feed überhaupt gibt, konsumieren nachgelagerte Jobs wie Preisüberwachung einen Stream statt eine Tabelle erneut zu lesen.

Die Abrufstufe bekommt die Aufmerksamkeit, weil dort die Blockaden und die Rechnungen entstehen. In der Speicherstufe entscheiden Sie, wie viel Abrufen Sie überhaupt bezahlen müssen.