Scraper API Webhooks vs. Polling: Aufbau einer zuverlässigen Ergebnis-Pipeline

Asynchrone Scraping-Aufträge werden nach ihrem eigenen Zeitplan abgeschlossen, und die Frage ist, wie Ihr Code dies erkennt. In diesem Leitfaden werden Web-Scraping-Webhooks mit Statusabfragen verglichen und anschließend die erforderlichen Komponenten für beide Ansätze erstellt: eine Abfrageschleife mit Backoff und einer festen Obergrenze, ein FastAPI-Empfänger, der Signaturen überprüft und Aufgaben in die Warteschlange stellt, eine Idempotenz-Tabelle, die Wiedergabe aus dem Zustellungsprotokoll sowie ein Fallback-Poller, der alles auffängt, was ein Webhook übersehen hat. Am Ende verfügen Sie über eine Ergebnispipeline, die keine Aufträge verliert oder doppelt verarbeitet.

Synchrone vs. asynchrone Extraktion

POST /v1/extract auf dem Geonode Scraper API läuft in einem von zwei Modi. Bei processing_mode: "sync" wird die Anfrage blockiert und gibt bei data.markdown oder data.html den Status 200 zurück, bei metadata (url, http_status, duration_ms) sowie tokens_charged zurück. Bei processing_mode: "async" wird der Status 202 zurückgegeben, bei job_id, status: "queued", status_url und estimated_tokens läuft der Auftrag im Hintergrund.

Sync ist die richtige Wahl, wenn Sie die Antwort direkt benötigen: bei einem Agenten-Tool-Aufruf, einem Request-Handler oder einem einmaligen Skript. Async ist die richtige Wahl, wenn die Seite JavaScript-Rendering und eine lange Wartezeit erfordert, wenn Sie viele URLs übermitteln oder wenn der Aufrufer die Verbindung nicht offen halten soll. Batch (POST /v1/batch) und Crawl (POST /v1/crawl) sind immer Jobs. Sobald Sie einen job_id haben, müssen Sie sich zwischen Polling- und Web-Scraping-Webhooks entscheiden.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

def submit(url: str) -> str:
    r = httpx.post(
        f"{BASE}/v1/extract",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "render_js": True,
            "processing_mode": "async",
        },
        timeout=30.0,
    )
    r.raise_for_status()  # 202 on success
    return r.json()["job_id"]

Polling: für wenige Aufträge geeignet, für viele jedoch kostspielig

Polling bedeutet, GET /v1/extract/{job_id} so lange aufzurufen, bis status den Status completed oder failed annimmt. Die Anleitung zum Auftragsstatus listet vier Zustände auf: queued, processing, completed, failed. Der Fehler, den die meisten Leute machen, ist eine feste Wartezeit von einer Sekunde. Das überlastet die API, während der Auftrag in der Warteschlange steht, und bei einer 429-Ratenbegrenzung (error.code "RATE_LIMITED") scheitert die Abfrageschleife, noch bevor der Auftrag fehlschlägt.

Verwenden Sie einen exponentiellen Backoff mit vollem Jitter und einer festen Obergrenze für die Gesamtwartezeit. Der Jitter verteilt gleichzeitige Abfragen so, dass sie nicht mehr alle in derselben Sekunde anstehen.

import os
import random
import time
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

class JobTimeout(Exception):
    pass

def poll_job(job_id: str, max_wait: float = 300.0) -> dict:
    deadline = time.monotonic() + max_wait
    delay = 1.0
    while True:
        r = httpx.get(f"{BASE}/v1/extract/{job_id}", headers=HEADERS, timeout=15.0)
        if r.status_code == 429:
            retry_after = float(r.headers.get("Retry-After", delay))
            time.sleep(retry_after)
            continue
        r.raise_for_status()
        job = r.json()
        if job["status"] in ("completed", "failed"):
            return job
        if time.monotonic() >= deadline:
            raise JobTimeout(job_id)
        # full jitter: sleep somewhere in [0, delay], then grow delay up to 30 s
        time.sleep(random.uniform(0, delay))
        delay = min(delay * 2, 30.0)

Das ist für ein Skript, das zehn URLs übermittelt, in Ordnung, für einen nächtlichen Batch von zwanzigtausend jedoch ungeeignet. Zweihundert laufende Jobs, die alle paar Sekunden abfragen, bedeuten Tausende leerer Statusabfragen pro Minute, und die Latenz wird weiterhin durch das Abfrageintervall begrenzt.

Wann Polling und wann Webhooks die Nase vorn haben

Webhooks verlagern die Komplexität von der Client-Schleife in Ihre Infrastruktur. Wenn Sie noch keinen öffentlichen Endpunkt und keine Warteschlange betreiben, ist das Abfragen die richtige erste Variante. Wenn Sie dies bereits tun, sind Web-Scraping-Webhooks kostengünstiger, schneller und skalieren mit Abschlussereignissen statt mit Sekunden.

Einen Webhook registrieren

Die Webhook-Endpunkte sind in der API-Referenz aufgeführt. POST /v1/webhooks akzeptiert url (erforderlich, eine URI), event_type (erforderlich, entweder extract_completed, batch_completed, crawl_completed) sowie einen optionalen description. Die Antwort 201 gibt den Webhook id, url, description, is_active, event_type, Zeitstempel sowie ein secret zurück. Das Geheimnis wird bei der Erstellung und durch POST /v1/webhooks/{webhook_id}/rotate-secret zurückgegeben; speichern Sie es zu diesem Zeitpunkt in einem Geheimnismanager, da Sie es zur Überprüfung jeder Zustellung verwenden werden.

Beachten Sie, dass die Webhook-Endpunkte zum Zeitpunkt der Erstellung dieses Dokuments zwar als dokumentiert, aber noch nicht in der Produktion verfügbar gekennzeichnet sind. Der untenstehende Vertrag entspricht der veröffentlichten Referenz; überprüfen Sie ihn vor der Bereitstellung.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

def register_webhook(receiver_url: str, event_type: str = "extract_completed") -> dict:
    r = httpx.post(
        f"{BASE}/v1/webhooks",
        headers=HEADERS,
        json={
            "url": receiver_url,
            "event_type": event_type,
            "description": "results pipeline, prod",
        },
        timeout=15.0,
    )
    r.raise_for_status()  # 201
    hook = r.json()
    # hook["secret"] is only shown here and after rotate-secret: persist it now
    return hook

Ein Webhook abonniert einen event_type, daher registriert eine Pipeline, die Extraktion, Batch-Verarbeitung und Crawling nutzt, drei davon.

Ein minimaler Empfänger, der die Signatur überprüft und das Ereignis in eine Warteschlange stellt

Ein Webhook-Empfänger hat genau drei Aufgaben: die Signatur zu überprüfen, so schnell wie möglich einen 2xx-Status zurückzugeben und das Ereignis an eine Warteschlange zu übergeben. Alles, was im Request-Handler länger als ein paar hundert Millisekunden dauert, ist ein Designfehler, da der Zustellversuch der Plattform möglicherweise in Zeitüberschreitung gerät und Sie Wiederholungsversuche für bereits erledigte Aufgaben erhalten.

Da in der Referenz der Name des Signatur-Headers, das Hash-Verfahren und das Payload-Schema noch nicht veröffentlicht sind, lässt der unten stehende Empfänger diese konfigurierbar und überprüft einen HMAC-SHA256 über den Rohtext des Body – was dem gängigen Muster entspricht. Überprüfen Sie den genauen Header und den Algorithmus in der Dokumentation und legen Sie diese über Umgebungsvariablen fest.

import hashlib
import hmac
import json
import os
import sqlite3

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

WEBHOOK_SECRET = os.environ["GEONODE_WEBHOOK_SECRET"].encode()
SIGNATURE_HEADER = os.environ.get("GEONODE_SIGNATURE_HEADER", "X-Webhook-Signature")  # confirm header name in docs
DB = sqlite3.connect("events.db", check_same_thread=False)
DB.execute(
    "CREATE TABLE IF NOT EXISTS processed_events ("
    "  job_id TEXT PRIMARY KEY,"
    "  event_type TEXT NOT NULL,"
    "  delivery_id TEXT,"
    "  received_at TEXT DEFAULT CURRENT_TIMESTAMP"
    ")"
)

def verify_signature(raw_body: bytes, header_value: str | None) -> bool:
    if not header_value:
        return False
    expected = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()  # confirm algorithm in docs
    return hmac.compare_digest(expected, header_value.strip())

def enqueue(event: dict) -> None:
    # Replace with your queue client (Redis, SQS, Celery, etc.).
    print("enqueued", event.get("job_id"))

@app.post("/webhooks/geonode")
async def receive(request: Request):
    raw = await request.body()
    if not verify_signature(raw, request.headers.get(SIGNATURE_HEADER)):
        raise HTTPException(status_code=401, detail="bad signature")

    event = json.loads(raw)
    job_id = event.get("job_id")  # confirm field name in docs
    event_type = event.get("event_type")  # confirm field name in docs
    delivery_id = event.get("id")  # confirm field name in docs
    if not job_id:
        raise HTTPException(status_code=400, detail="missing job_id")

    try:
        DB.execute(
            "INSERT INTO processed_events (job_id, event_type, delivery_id) VALUES (?, ?, ?)",
            (job_id, event_type, delivery_id),
        )
        DB.commit()
    except sqlite3.IntegrityError:
        # Duplicate delivery: already recorded, acknowledge and do nothing.
        return {"status": "duplicate"}

    enqueue({"job_id": job_id, "event_type": event_type})
    return {"status": "accepted"}

Drei Details sind entscheidend. Die Signatur wird über die rohen Bytes berechnet, nicht über ein neu serialisiertes JSON-Objekt, da jede Änderung der Leerzeichen die Signatur ungültig macht. Das Einfügen in processed_events erfolgt vor dem Einreihen in die Warteschlange; ein Absturz zwischen diesen beiden Schritten hinterlässt daher eine Zeile ohne Bearbeitung, die später vom Fallback-Poller korrigiert wird. Und der Handler gibt bei einem Duplikat einen 2xx-Status zurück statt eines Fehlers, da ein Fehler die Plattform anweist, bereits übermittelte Daten erneut zu versuchen.

Idempotenz und nicht in der richtigen Reihenfolge erfolgte Zustellungen

Jedes Webhook-System arbeitet nach dem „At-Least-Once“-Prinzip. Netzwerkstörungen, ein langsamer Worker, ein rollierender Deployment, bei dem eine Antwort verloren geht: All dies führt zu einer zweiten Zustellung für denselben Auftrag. Die Tabelle „processed_events“ mit „job_id“ als Primärschlüssel ist die Lösung des Problems. Eine Eindeutigkeitsbeschränkung macht die Duplikatsprüfung atomar, was bei einem „SELECT“, gefolgt von einem „INSERT“, nicht der Fall ist.

Eine falsche Reihenfolge ist ein anderes Problem. Ein batch_completed-Ereignis kann eintreffen, bevor Sie die Erfassung der Batch-Übermittlung abgeschlossen haben, oder zwei Ereignisse für miteinander verbundene Aufträge können in umgekehrter Reihenfolge eintreffen. Der Worker sollte nicht von der Reihenfolge abhängig sein. Die einzige Eingabe für den Worker sollte ein „job_id“ sein; er ruft den aktuellen Status des Jobs über die API ab und handelt entsprechend – dieser Status ist immer korrekt, unabhängig davon, wann das Ereignis eingetroffen ist.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

def store_result(job_id: str, job: dict) -> None: ...   # write to your store
def resubmit(job_id: str) -> None: ...                  # re-POST the original URL
def dead_letter(job_id: str, err: dict) -> None: ...    # park for manual review

def handle_job(job_id: str) -> None:
    r = httpx.get(f"{BASE}/v1/extract/{job_id}", headers=HEADERS, timeout=15.0)
    r.raise_for_status()
    job = r.json()
    if job["status"] == "completed":
        store_result(job_id, job)
    elif job["status"] == "failed":
        err = job.get("error", {})  # confirm field name in docs
        if err.get("retryable"):
            resubmit(job_id)
        else:
            dead_letter(job_id, err)
    else:
        # Event arrived before the job settled; let the fallback poller retry.
        pass

Nachverfolgung aus dem Lieferprotokoll

GET /v1/webhooks/{webhook_id}/deliveries listet jeden Versuch auf, den die Plattform unternommen hat. Jeder Eintrag enthält id, event_type, job_id, attempt_number, status, http_status, delivered_at und created_at, wobei page, page_size (max. 100) und ein optionaler Filter status als Abfrageparameter dienen. Die möglichen Werte für status sind in der Referenz nicht aufgeführt. Filtern Sie daher stattdessen nach http_status, wenn Sie Lieferungen finden möchten, die Ihr Endpunkt abgelehnt hat.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

def replay_failed(webhook_id: str) -> list[str]:
    missed: list[str] = []
    page = 1
    while True:
        r = httpx.get(
            f"{BASE}/v1/webhooks/{webhook_id}/deliveries",
            headers=HEADERS,
            params={"page": page, "page_size": 100},
            timeout=15.0,
        )
        r.raise_for_status()
        body = r.json()
        for d in body["items"]:
            ok = d.get("http_status") is not None and 200 <= d["http_status"] < 300
            if not ok and d["job_id"] not in missed:
                missed.append(d["job_id"])
        if page >= body["page_count"]:
            break
        page += 1
    for job_id in missed:
        handle_job(job_id)
    return missed

Da handle_job dieselbe Idempotenz-Tabelle durchläuft, ist die Wiederholung eines Jobs, der tatsächlich erfolgreich war, unbedenklich.

Der Fallback-Poller

Das Zustellungsprotokoll informiert Sie über Zustellversuche der Plattform. Es liefert jedoch keine Informationen zu Ereignissen, die nie generiert wurden, oder zu einem deaktivierten Webhook. Die letzte Verteidigungslinie ist eine langsame Abgleichschleife: Listen Sie alle paar Minuten die Jobs mit GET /v1/extract/jobs auf, und verarbeiten Sie jeden Job, der älter als Ihr Schwellenwert ist und für den keine Zeile in processed_events vorhanden ist, direkt.

import os
import httpx
from datetime import datetime, timedelta, timezone

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

def reconcile(older_than: timedelta = timedelta(minutes=10)) -> None:
    r = httpx.get(f"{BASE}/v1/extract/jobs", headers=HEADERS, timeout=30.0)
    r.raise_for_status()
    cutoff = datetime.now(timezone.utc) - older_than
    for job in r.json().get("items", []):  # confirm field name in docs
        created = datetime.fromisoformat(job["created_at"].replace("Z", "+00:00"))  # confirm field name in docs
        if created > cutoff or job["status"] not in ("completed", "failed"):
            continue
        seen = DB.execute(
            "SELECT 1 FROM processed_events WHERE job_id = ?", (job["job_id"],)
        ).fetchone()
        if seen is None:
            DB.execute(
                "INSERT INTO processed_events (job_id, event_type) VALUES (?, ?)",
                (job["job_id"], "reconciled"),
            )
            DB.commit()
            handle_job(job["job_id"])

Führen Sie dies in einem Rhythmus aus, der langsam genug ist, um kostengünstig zu sein, und schnell genug, damit ein ausgefallener Webhook nicht zu einer Datenlücke führt, die Ihren Benutzern auffällt. Zehn Minuten sind ein sinnvoller Ausgangspunkt.

Rotation des Geheimnisses

Rotieren Sie das Geheimnis nach einem festgelegten Zeitplan und sofort, falls es bekannt wird. POST /v1/webhooks/{webhook_id}/rotate-secret wird durch ein neues secret ersetzt, wodurch das vorherige gemäß der Dokumentation ungültig wird. Das bedeutet, dass es zwischen dem API-Wechsel und dem Neuladen der Konfiguration durch Ihren Empfänger ein Zeitfenster gibt, in dem die Übermittlungen die Überprüfung nicht bestehen. Halten Sie dieses Zeitfenster kurz: Lassen Sie den Empfänger den geheimen Schlüssel aus einer Quelle lesen, die er ohne Neustart aktualisieren kann, rotieren Sie den Schlüssel, aktualisieren Sie die Quelle und überprüfen Sie anschließend das Übermittlungsprotokoll auf Einträge mit dem Status 401, um diese erneut zu versenden.

Häufige Fehler

Ausführung der Verarbeitung innerhalb des Request-Handlers

Das Abrufen des Ergebnisses, dessen Auswertung und das Schreiben in die Datenbank innerhalb des Webhook-Handlers ist der häufigste Fehler. Die Übermittlungszeit der Plattform läuft ab, es erfolgt ein erneuter Versuch, und schon haben Sie Duplikate und einen langsamen Endpunkt. Bestätigen und in die Warteschlange stellen – mehr nicht.

Auf Duplikate prüfen ohne Eindeutigkeitsbeschränkung

Bei einer Abfolge von SELECT und INSERT kommt es zu einem Wettlauf zwischen zwei gleichzeitigen Übermittlungen desselben Auftrags. Machen Sie job_id zum Primärschlüssel und lassen Sie die Datenbank den zweiten Eintrag ablehnen.

Der Ereignis-Payload gegenüber dem Jobstatus vertrauen

Behandeln Sie den Webhook als Benachrichtigung, nicht als Daten. Der Worker sollte GET /v1/extract/{job_id} immer erneut abrufen und auf das echte status reagieren, was auch die Wiederholung und den Abgleich trivial macht.

Ausführung von Webhooks ohne Abgleich

Web-Scraping-Webhooks funktionieren nach dem „At-Least-Once“-Prinzip, nicht nach „Exactly-Once“, und es ist nicht garantiert, dass sie bei jedem Ausfall, den Sie haben werden, mindestens einmal ausgeführt werden. Ohne einen Fallback-Poller werden Sie irgendwann eine Lücke entdecken, ohne wissen zu können, wie groß diese ist.

Nächster Schritt

Alles oben Genannte läuft auf dem Geonode Scraper API: asynchrone Extraktion mit processing_mode: "async", Batch- und Crawling-Jobs sowie Webhook-Abonnements mit rotierbaren Geheimnissen und einem abfragbaren Zustellungsprotokoll. Es gibt eine kostenlose Stufe mit 1.500 Anfragen pro Monat, um den Empfänger aufzubauen und zu testen, und pay-as-you-go ab 3,50 $ pro 10.000 Anfragen darüber hinaus. Beginnen Sie mit Polling, fügen Sie Webhooks hinzu, wenn die Anzahl der Jobs dies rechtfertigt, und lassen Sie den Reconciler in jedem Fall weiterlaufen.

FAQ

Was sind Web-Scraping-Webhooks und wann sollte ich sie verwenden?

Web-Scraping-Webhooks sind HTTP-Callbacks, die die Scraper-Plattform an Ihren Server sendet, sobald ein asynchroner Job abgeschlossen ist, sodass Sie nicht ständig nach dessen Status fragen müssen. Verwenden Sie sie, wenn Sie Batch- oder Crawling-Jobs ausführen oder mehr als ein paar Dutzend asynchrone Extraktionen gleichzeitig durchführen. Bei einer Handvoll kurzer Jobs ist Polling einfacher und völlig ausreichend.

Ist das Polling eines scraper API eine schlechte Praxis?

Nein. Polling ist die richtige Wahl für kleine Datenmengen, Skripte, Notebooks und alles, wo eine Latenz von wenigen Sekunden keine Rolle spielt. Es wird erst dann zum Problem, wenn Hunderte von Jobs gleichzeitig laufen, da jeder einzelne pro Polling-Intervall eine Anfrage kostet und Sie dennoch nicht schneller als dieses Intervall reagieren können. Hier zahlen sich Web-Scraping-Webhooks aus.

Wie kann ich überprüfen, ob ein Webhook wirklich von der Scraper-Plattform stammt?

Jeder Geonode-Webhook gibt bei seiner Erstellung ein Signaturgeheimnis zurück, das regelmäßig gewechselt werden kann. Der Empfänger berechnet mit diesem Schlüssel eine neue Signatur über den Rohtext des Anfrage-Hauptteils und vergleicht diese in konstanter Zeit mit dem Header, den die Plattform sendet. Lehnen Sie alles ab, was die Prüfung nicht besteht, bevor Sie die Nutzdaten auswerten, und protokollieren Sie den Schlüssel niemals.

Was passiert, wenn mein Webhook-Endpunkt bei Abschluss eines Jobs nicht erreichbar ist?

Die Plattform protokolliert einen Zustellversuch, den Sie über GET /v1/webhooks/{webhook_id}/deliveries einsehen können; dort werden attempt_number, status und http_status pro Versuch aufgelistet. Ihre Pipeline sollte sich nicht allein auf eine erneute Zustellung verlassen. Führen Sie einen Fallback-Poller aus, der alle Jobs, die älter als ein Schwellenwert sind und für die noch kein Ereignis verarbeitet wurde, abgleicht und deren Ergebnis direkt abruft.

Sind die Webhooks Geonode und Scraper API allgemein verfügbar?

Zum Zeitpunkt der Erstellung dieses Artikels sind die Webhook-Endpunkte auf docs.geonode.com dokumentiert, jedoch als in der Produktion noch nicht verfügbar gekennzeichnet, und der Signatur-Header, das Payload-Schema sowie der Wiederholungszeitplan sind nicht veröffentlicht. Erstellen Sie den Empfänger gemäß dem dokumentierten Vertrag, halten Sie Header- und Feldnamen konfigurierbar und überprüfen Sie diese anhand der Dokumentation, bevor Sie das Produkt ausliefern.

Benötige ich sowohl Web-Scraping-Webhooks als auch Polling?

In der Produktion ja. Webhooks bieten Ihnen eine geringe Latenz und ein geringes Anforderungsvolumen. Ein langsamer Abgleich-Poller erfasst Aufträge, deren Zustellung fehlgeschlagen ist, deren Endpunkt abgelehnt wurde oder die aus Ihrer Warteschlange gefallen sind. Zusammen gewährleisten sie eine „At-Least-Once“-Verarbeitung, und eine auf der job_id basierende Idempotenz-Tabelle wandelt dies in eine effektive „Once“-Verarbeitung um.