SQLAlchemy para web scraping: uma camada de armazenamento de produção

Resposta curta: em um pipeline de scraping, o SQLAlchemy não é a parte que coleta — é a parte que torna a nova coleta desnecessária. Modele o registro de que você realmente precisa, limite o pool de conexões abaixo do limite do próprio banco, grave cada linha por meio de um ON CONFLICT DO UPDATE para que uma retentativa não custe nada, e repita a transação inteira em vez do comando. Tudo o que segue tem como alvo o SQLAlchemy 2.0 — 2.0.52, lançada em 11 de agosto de 2026 — e cita os valores padrão oficiais em vez de adivinhá-los.

Onde o banco de dados entra em um pipeline de scraping

Um pipeline de produção tem cinco etapas — obtenção de URLs, coleta, parsing, validação e armazenamento — e nosso guia de web scraping percorre a cadeia inteira. Só uma dessas etapas custa dinheiro por byte. O tráfego de proxies residenciais é cobrado por gigabyte, então a economia do pipeline é decidida pela frequência com que você precisa coletar uma página uma segunda vez; nossa estimativa de custos de web scraping coloca números reais nisso.

Isso reposiciona a camada de armazenamento. O trabalho dela não é «guardar as linhas em algum lugar». É responder a três perguntas de forma barata o bastante para que o coletor nunca precise:

1. Já vi este registro, e ele mudou? Se a resposta for não, a coleta economiza uma requisição cobrada.
2. O que sobrevive a uma queda? Um worker que morre no meio do lote não pode deixar metade de uma página confirmada e a outra metade perdida.
3. Os analistas conseguem ler enquanto o crawler grava? Uma transação interminável segurando bloqueios transforma o data warehouse e o crawler em inimigos.

O SQLAlchemy responde às três, em duas camadas que vale a pena não confundir. O Core é a linguagem de expressão SQL: Table, select(), insert(), o motor e o pool de conexões. O ORM acrescenta sobre isso classes mapeadas, um identity map e a Session como unidade de trabalho. A divisão produtiva para um scraper é definir o esquema com modelos declarativos do ORM — assim os tipos do Python e o DDL têm uma única fonte de verdade — e gravar lotes com comandos no estilo Core, porque inserir dez mil linhas é uma operação de conjunto, não dez mil mutações de objeto.

Modele o registro, não a página que você raspou

O erro de modelagem mais caro em um scraper é guardar o artefato em vez do fato. HTML bruto é enorme, muda por motivos que não interessam e não pode ser consultado. Guarde o registro extraído mais os metadados suficientes para decidir se ele é novo.

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

Três detalhes ali fazem a maior parte do trabalho.

Mapped[] decide a nulidade. A regra documentada é que Mapped[str] produz NOT NULL e Mapped[Optional[str]] — ou Mapped[str | None] — produz NULL, com primary_key=True implicando NOT NULL de qualquer forma. Você ainda pode forçar com mapped_column(nullable=...), mas, no padrão, o verificador de tipos lê a mesma verdade que o esquema.

type_annotation_map centraliza o mapeamento de Python para SQL. O SQLAlchemy já mapeia int para Integer, str para String, datetime.datetime para DateTime, uuid.UUID para Uuid e assim por diante. Sobrescrever esse mapa na classe base é como se aplica uma política de projeto inteiro — carimbos de tempo com fuso, JSONB em vez do JSON genérico — sem repeti-la em 40 colunas.

A chave natural é uma restrição real, não uma convenção. source mais external_id é o que torna a mesma página de produto reconhecível entre coletas. Um id substituto sozinho não expressa isso, e tudo na próxima seção depende de o banco conseguir detectar a colisão sozinho.

O motor e o pool são o seu orçamento de concorrência

create_engine() é uma fábrica, não uma conexão: ele mantém um pool de conexões criado de forma preguiçosa e compartilhado. Fora o SQLite em memória, o pool padrão é o QueuePool, e os padrões documentados dele são os números que vão te morder primeiro.

A aritmética que importa é workers × (pool_size + max_overflow) ≤ limite de conexões − folga. Com os padrões, oito processos de coleta podem exigir 120 conexões de um banco que talvez permita 100 no total, e a falha aparece como pool_timeout estourando nos workers enquanto uma migração sequer consegue conectar. Decida o teto de forma deliberada:

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 é a estratégia pessimista documentada contra desconexões: o SQLAlchemy envia na retirada um ping específico do dialeto (normalmente SELECT 1) e, se ele falhar, descarta aquela conexão e invalida todas as conexões do pool mais antigas que o instante atual. Custa uma ida e volta por retirada e elimina toda a classe «a primeira consulta depois de um período ocioso falha». O pool_recycle é o complemento para bancos que fecham conexões ociosas por temporizador — a documentação o aponta como remédio imediato para o server has gone away do MySQL, e o mesmo raciocínio vale para qualquer PostgreSQL gerenciado atrás de um proxy de conexões.

O fork é onde os pools quebram

Crawlers com multiprocessamento esbarram em um perigo específico e documentado: as conexões do pool não são compartilhadas com um processo bifurcado. Dois processos acabam gravando pelo mesmo socket, e o sintoma é estado de protocolo corrompido em vez de um erro limpo. O remédio oficial é descartar o pool herdado no filho, sem fechar os sockets do pai:

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)

Uma sessão por unidade de trabalho

A Session é documentada como «a mutable, stateful object that represents a single database transaction», e «cannot be shared among concurrent threads or asyncio tasks without careful synchronization». Leia isso como regra de projeto, não como aviso: a vida de uma sessão é a de uma transação, que é o tempo em que você segura bloqueios.

Crie o sessionmaker uma vez no escopo do módulo e abra uma sessão por unidade de trabalho:

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() como gerenciador de contexto confirma em caso de sucesso e desfaz em qualquer exceção, então não existe caminho pelo qual um lote pela metade escape. Lotes de cerca de 200 a 1.000 linhas são o ponto ideal habitual: grandes o bastante para que as idas e voltas deixem de dominar, pequenos o bastante para que desfazer seja barato e a duração dos bloqueios fique em dezenas de milissegundos.

Torne toda gravação idempotente

Esta é a seção que justifica o artigo. Um crawler é um sistema que vai rodar de novo: retentativas, reprocessamentos, um operador reiniciando o job de ontem. Se gravar a mesma página duas vezes gera duplicatas ou destrói dados bons, cada um desses eventos vira um incidente. INSERT ... ON CONFLICT DO UPDATE move a deduplicação para o banco, onde ela é atômica.

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

Repare no caminho de importação: o upsert vive no insert específico do dialeto, insert de sqlalchemy.dialects.postgresql, não no genérico. O SQLite oferece a mesma grafia on_conflict_do_update(), MySQL e MariaDB oferecem on_duplicate_key_update(), e não existe construção portável que cubra os três. É uma razão concreta para desenvolver no motor que você implanta.

A cláusula where merece um momento de honestidade sobre o trade-off. Pular a atualização significa que last_changed_at registra quando o conteúdo mudou pela última vez, que é exatamente o que um pipeline de detecção de mudanças quer, mas você deixa de registrar quando conferiu pela última vez. Se precisar dos dois, mantenha a atualização protegida para o registro em si e grave o carimbo barato de «visto em» numa tabela estreita separada, com a qual ninguém faz join.

A recompensa é um feed de mudanças de graça. Como as linhas inalteradas são puladas, o RETURNING devolve exatamente as que se moveram:

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)

Essa é a diferença entre «reindexamos tudo toda noite» e «reindexamos os 0,4% que mudaram». Num corpus de um milhão de páginas, é a diferença entre um job de minutos e um que ocupa a noite inteira.

Gravações em massa: deixe o insertmanyvalues agrupar

O SQLAlchemy 2.0 aceita uma lista de dicionários como conjunto de parâmetros de um insert(), e o ORM interpreta as chaves como nomes de atributo em vez de nomes de coluna — uma mudança deliberada da 2.0, que importa no momento em que um atributo mapeado e sua coluna são escritos de formas diferentes.

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

Por baixo, o recurso insertmanyvalues reescreve isso em comandos INSERT ... VALUES agrupados de várias linhas. Ele vem ligado por padrão para PostgreSQL, MySQL, SQLite, SQL Server e Oracle, e o tamanho do lote segue insertmanyvalues_page_size, que «defaults to 1000, but may also be subject to dialect-specific limiting factors». Baixe o valor quando as linhas carregarem grandes cargas JSON: o ganho vem de menos idas e voltas, e um comando grande demais para os limites de parâmetros do servidor devolve esse ganho na hora.

Quando precisar das chaves primárias geradas, peça-as na ordem dos parâmetros:

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

Duas coisas a não fazer. Não percorra session.add() em dez mil objetos para confirmar uma vez: você paga toda a contabilidade da unidade de trabalho por linhas que ninguém vai alterar. E não monte o SQL como string para «evitar a sobrecarga do ORM»; o caminho parametrizado é justamente o que mantém o texto raspado — entrada não confiável por definição — fora da sua gramática SQL.

Pipelines assíncronos: uma AsyncSession por tarefa

Se o seu coletor já é asyncio — a maioria dos modernos é, e nossas notas sobre proxies para agentes de IA explicam por que esse formato se espalhou — rodar o banco no mesmo laço de eventos evita um salto de thread. As regras são estreitas e inegociáveis.

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

O erro que você encontra ao errar chama-se MissingGreenlet, e quase sempre significa um carregamento preguiçoso disparado fora do contexto assíncrono. A documentação de asyncio oferece três saídas, na ordem em que eu tentaria: carregar antecipadamente com selectinload() na consulta, acrescentar o mixin AsyncAttrs à sua base e escrever await obj.awaitable_attrs.things, ou descer para session.run_sync() num bloco de código ORM síncrono comum.

Um alerta sobre dimensionamento: o assíncrono não eleva o limite de conexões do seu banco — só torna muito mais fácil alcançá-lo. Mil tarefas concorrentes contra pool_size=10 não são um impasse, mas toda tarefa depois da décima entra na fila, e pool_timeout decide se essa fila falha em alto e bom som ou vira silenciosamente a sua latência.

Ler de volta sem N+1

O caminho de leitura é onde os dados raspados são exportados, e onde mora o clássico defeito de desempenho do ORM: iterar sobre pais tocando um relacionamento a cada volta emite uma consulta por pai. A resposta do SQLAlchemy é declarar a estratégia de carregamento na consulta.

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

Para um relacionamento escalar muitos-para-um — uma observação de volta para a sua página — joinedload() é a forma melhor, porque o join acrescenta um conjunto de colunas em vez de uma segunda ida e volta. A única armadilha que a documentação declara explicitamente: joinedload() sobre uma coleção multiplica linhas, então os resultados precisam ser deduplicados com .unique() antes de qualquer contagem.

raiseload("*") é subestimado. Ligue-o nos testes e em qualquer exportação em lote e todo carregamento preguiçoso acidental vira uma falha barulhenta e localizada durante o desenvolvimento, em vez de um job misterioso de uma hora em produção.

Observabilidade: meça o SQL, não a impressão

echo=True é uma chave de desenvolvimento. Em produção ele registra todo comando — inclusive os parâmetros ligados, o que num scraper significa conteúdo raspado e, na própria conexão, credenciais. Use eventos e registre o comando sem os parâmetros.

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)

A assinatura do evento é fixa — (conn, cursor, statement, parameters, context, executemany) para os dois hooks — e a pilha em conn.info mantém execuções aninhadas corretamente pareadas. Três métricas valem a exportação: duração por tipo de operação, a fatia de executemany (que sobe quando seu agrupamento funciona) e o tempo de espera na retirada do pool, o número que diz se o crawler está lento por causa do site alvo ou do seu próprio pool_size.

Repita a transação, nunca o comando

A documentação é direta: «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». A forma recomendada é «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))

Isso só é seguro porque a gravação é um upsert. Repetir um INSERT puro depois de uma falha ambígua é como se produzem duplicatas; repetir um ON CONFLICT DO UPDATE converge para a mesma linha por mais vezes que rode. Aqui a idempotência não é um luxo: é a pré-condição que torna a recuperação automática possível.

pool_pre_ping e esse decorador resolvem metades diferentes do problema. O pre-ping pega a conexão que morreu ociosa no pool, antes de você começar o trabalho. A retentativa pega a conexão que morre no meio da transação, quando não há nada a salvar.

Mudanças de esquema: autogenerate é rascunho, não migração

O Alembic é a ferramenta de migração do mesmo projeto, e a própria documentação dele diz que o autogenerate «is not intended to be perfect» e que «it is always necessary to manually review and correct the candidate migrations that autogenerate produces». Leve isso ao pé da letra.

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

Ele detecta com segurança adições e remoções de tabelas e colunas, mudanças de nulidade, mudanças básicas em índices e restrições únicas nomeadas, mudanças básicas em chaves estrangeiras e restrições CHECK nomeadas. Não detecta renomeações de tabela nem de coluna — ambas aparecem como remoção mais adição, o que num corpus raspado significa apagar a coluna e reconstruí-la vazia — e não enxerga restrições com nome anônimo. É por isso que a naming_convention está no MetaData do primeiro bloco de código: sem ela, o autogenerate não tem como parear as restrições.

O fluxo que se sustenta: gerar, ler o arquivo linha a linha, reescrever os pares remoção-adição como op.alter_column(..., new_column_name=...) e rodar a migração contra uma cópia restaurada da produção antes que ela chegue perto da produção.

Testes: desfaça em vez de limpar

Testes de scraper precisam de um banco que se comporte como o real e de uma fixture que não deixe resíduo. O SQLAlchemy 2.0 documenta o padrão: abrir uma transação externa numa conexão, ligar a sessão a essa conexão em modo savepoint e desfazer a transação externa no fim do teste.

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

Rode isso no mesmo motor que você implanta. O SQLite é um alvo válido para as partes de Python puro, mas a semântica de ON CONFLICT, os operadores JSONB e a ordenação de NULL são todos diferentes, e uma suíte que passa no SQLite enquanto a produção roda Postgres está testando outro programa.

Cinco falhas que sempre encontro em bancos de scrapers

Nada disso é exótico. São as mesmas cinco todas as vezes, e quatro das cinco são configuração, não código.

O que isso entrega ao lado da coleta

Uma camada de armazenamento construída assim muda o que o crawler pode fazer. Como as gravações são idempotentes, um worker pode cair e reiniciar sem reconciliação. Como o upsert relata o que mudou, o agendador pode coletar com mais frequência as páginas que mudam rápido e com menos frequência as calmas — a maior alavanca sobre o gasto com proxies, e que se encaixa direto nas práticas de como raspar sem ser bloqueado. E porque o feed de mudanças existe, jobs a jusante como o monitoramento de preços consomem um fluxo em vez de reler uma tabela.

A etapa de coleta recebe a atenção porque é onde estão os bloqueios e as faturas. A etapa de armazenamento é onde você decide quanta coleta vai precisar pagar.