Tutorials

DynamoDB für serverlose CAPTCHA-Lösungsverfolgung

Automatisierung Python Alle CAPTCHA-Typen
Von CaptchaAI Team Deutsch
Veröffentlicht unter Jun 25, 2026
DynamoDB für serverlose CAPTCHA-Lösungsverfolgung
Ausgewähltes Bild nicht verfügbar

Wenn jeder Lambda-Aufruf eine neue Datenbankverbindung öffnet, wird eine klassische SQL-Datenbank schnell zum Flaschenhals. Für die Protokollierung von CAPTCHA-Lösungen in serverlosen Workflows ist DynamoDB deshalb die naheliegende Wahl: kein Verbindungspooling, integriertes TTL für die automatische Bereinigung und gleichbleibende Latenz unabhängig vom Durchsatz. Dieser Leitfaden zeigt Tabellendesign, Elementstruktur und Abfragemuster, mit denen Sie jeden Solve von CaptchaAI – Zeitstempel, Dauer, Status und CAPTCHA-Typ – nachvollziehbar ablegen und später auswerten.

Der Aufbau folgt vier Schritten: zuerst das Datenmodell, dann die Python-Integration, anschließend dieselbe Logik in JavaScript und zum Schluss Kosten- und Betriebsfragen. Alle Codebeispiele nutzen die CaptchaAI-Endpunkte in.php und res.php und speichern das Ergebnis in einer einzigen DynamoDB-Tabelle.

Tabellendesign für das Solve-Tracking

Single-Table-Design in DynamoDB

Eine einzige Tabelle deckt drei Anwendungsfälle ab: den einzelnen Lösungsdatensatz, den Verlauf pro Website und die täglich aggregierte Statistik. Das spart Verwaltungsaufwand und passt zur DynamoDB-Philosophie, Zugriffsmuster über zusammengesetzte Schlüssel abzubilden statt über mehrere Tabellen mit Joins.

Partitionsschlüssel (PK) Sortierschlüssel (SK) Zweck
SOLVE#{captcha_id} META Einzelner Lösungsdatensatz
SITE#{sitekey} SOLVE#{timestamp} Lösungsverlauf pro Site
STATS#{date} TYPE#{captcha_type} Tägliche Aggregatstatistik
ACTIVE#{captcha_id} TASK Laufende Aufgabe in Bearbeitung

Der Präfix vor dem # gruppiert verwandte Elemente in derselben Partition. So liefert eine einzelne Abfrage auf SITE#{sitekey} den kompletten Verlauf einer Website, sortiert nach Zeitstempel – ganz ohne Scan über die gesamte Tabelle.

Tabellendefinition mit TTL und GSI

Die Tabelle nutzt PAY_PER_REQUEST (On-Demand), einen Global Secondary Index für statusbasierte Abfragen und ein TTL-Attribut, das abgelaufene Datensätze automatisch entfernt.

{
  "TableName": "CaptchaSolves",
  "KeySchema": [
    { "AttributeName": "PK", "KeyType": "HASH" },
    { "AttributeName": "SK", "KeyType": "RANGE" }
  ],
  "AttributeDefinitions": [
    { "AttributeName": "PK", "KeyType": "S" },
    { "AttributeName": "SK", "KeyType": "S" },
    { "AttributeName": "GSI1PK", "KeyType": "S" },
    { "AttributeName": "GSI1SK", "KeyType": "S" }
  ],
  "GlobalSecondaryIndexes": [
    {
      "IndexName": "GSI1",
      "KeySchema": [
        { "AttributeName": "GSI1PK", "KeyType": "HASH" },
        { "AttributeName": "GSI1SK", "KeyType": "RANGE" }
      ],
      "Projection": { "ProjectionType": "ALL" }
    }
  ],
  "BillingMode": "PAY_PER_REQUEST",
  "TimeToLiveSpecification": {
    "AttributeName": "ttl",
    "Enabled": true
  }
}

Python-Integration mit CaptchaAI

Setup: boto3 und API-Schlüssel

Die Konfiguration bleibt bewusst schlank. Tabellenname und API-Schlüssel kommen aus Umgebungsvariablen – so landet kein Geheimnis im Code, und dieselbe Lambda-Funktion läuft in Staging und Produktion unverändert.

import os
import time
from datetime import datetime, timezone
import boto3
import requests

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ.get("DYNAMODB_TABLE", "CaptchaSolves"))
API_KEY = os.environ["CAPTCHAAI_API_KEY"]

Lösen und protokollieren

Die zentrale Funktion übermittelt das CAPTCHA an CaptchaAI, legt sofort einen Eintrag für die laufende Aufgabe an und fragt das Ergebnis im Polling-Verfahren ab. Bei Erfolg schreibt sie den Lösungsdatensatz, ergänzt den Verlauf pro Site, entfernt die laufende Aufgabe und aktualisiert die Tagesstatistik. Fehler und Timeouts werden ebenso festgehalten – gerade diese Datensätze sind später für die Erfolgsquote entscheidend.

def solve_and_track(sitekey, pageurl, captcha_type="recaptcha_v2", project=None):
    now = datetime.now(timezone.utc)
    timestamp = now.isoformat()
    ttl_90_days = int(now.timestamp()) + (90 * 24 * 3600)

    # Submit to CaptchaAI
    resp = requests.post("https://ocr.captchaai.com/in.php", data={
        "key": API_KEY,
        "method": "userrecaptcha",
        "googlekey": sitekey,
        "pageurl": pageurl,
        "json": 1
    })
    data = resp.json()

    if data.get("status") != 1:
        # Store error record
        table.put_item(Item={
            "PK": f"SITE#{sitekey}",
            "SK": f"SOLVE#{timestamp}",
            "captcha_type": captcha_type,
            "pageurl": pageurl,
            "status": "error",
            "error": data.get("request"),
            "submitted_at": timestamp,
            "project": project or "default",
            "ttl": ttl_90_days,
            "GSI1PK": f"STATUS#error",
            "GSI1SK": timestamp
        })
        return {"error": data.get("request")}

    captcha_id = data["request"]

    # Track active task
    table.put_item(Item={
        "PK": f"ACTIVE#{captcha_id}",
        "SK": "TASK",
        "sitekey": sitekey,
        "pageurl": pageurl,
        "captcha_type": captcha_type,
        "submitted_at": timestamp,
        "ttl": int(now.timestamp()) + 600  # Auto-clean in 10 min
    })

    # Poll for result
    polls = 0
    for _ in range(60):
        time.sleep(5)
        polls += 1
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get",
            "id": captcha_id, "json": 1
        }).json()

        if result.get("status") == 1:
            solved_at = datetime.now(timezone.utc).isoformat()
            elapsed_ms = int(
                (datetime.now(timezone.utc) - now).total_seconds() * 1000
            )

            # Store success record
            table.put_item(Item={
                "PK": f"SOLVE#{captcha_id}",
                "SK": "META",
                "captcha_type": captcha_type,
                "sitekey": sitekey,
                "pageurl": pageurl,
                "status": "solved",
                "submitted_at": timestamp,
                "solved_at": solved_at,
                "elapsed_ms": elapsed_ms,
                "polls": polls,
                "project": project or "default",
                "ttl": ttl_90_days,
                "GSI1PK": f"STATUS#solved",
                "GSI1SK": timestamp
            })

            # Also store in site history
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "solved",
                "elapsed_ms": elapsed_ms,
                "ttl": ttl_90_days
            })

            # Remove active task
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })

            # Update daily stats
            update_daily_stats(captcha_type, True, elapsed_ms)

            return {"solution": result["request"]}

        if result.get("request") != "CAPCHA_NOT_READY":
            table.put_item(Item={
                "PK": f"SITE#{sitekey}",
                "SK": f"SOLVE#{timestamp}",
                "captcha_id": captcha_id,
                "status": "error",
                "error": result.get("request"),
                "ttl": ttl_90_days
            })
            table.delete_item(Key={
                "PK": f"ACTIVE#{captcha_id}", "SK": "TASK"
            })
            update_daily_stats(captcha_type, False, 0)
            return {"error": result.get("request")}

    table.delete_item(Key={"PK": f"ACTIVE#{captcha_id}", "SK": "TASK"})
    update_daily_stats(captcha_type, False, 0)
    return {"error": "TIMEOUT"}


def update_daily_stats(captcha_type, success, elapsed_ms):
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    update_expr = "SET total_solves = if_not_exists(total_solves, :zero) + :one"
    expr_values = {":zero": 0, ":one": 1}

    if success:
        update_expr += ", successful = if_not_exists(successful, :zero) + :one"
        update_expr += ", total_elapsed = if_not_exists(total_elapsed, :zero) + :elapsed"
        expr_values[":elapsed"] = elapsed_ms
    else:
        update_expr += ", failed = if_not_exists(failed, :zero) + :one"

    table.update_item(
        Key={"PK": f"STATS#{date_str}", "SK": f"TYPE#{captcha_type}"},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values
    )

Die Funktion update_daily_stats nutzt eine atomare UpdateExpression mit if_not_exists. Das ist wichtig, weil in einer serverlosen Umgebung viele Lambda-Aufrufe parallel dieselbe Statistik-Partition beschreiben – DynamoDB serialisiert die Zähler-Updates korrekt, ohne dass Sie eine Sperre selbst implementieren müssen.

Abfragemuster für Verlauf und Statistik

Drei Zugriffsmuster decken die meisten Auswertungen ab: der Verlauf pro Website, die Tageswerte pro CAPTCHA-Typ und die aktuell laufenden Aufgaben über den GSI.

def get_site_history(sitekey, limit=50):
    """Get recent solves for a specific site key."""
    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"SITE#{sitekey}"},
        ScanIndexForward=False,
        Limit=limit
    )
    return response["Items"]


def get_daily_stats(date_str=None):
    """Get stats for a specific date (default: today)."""
    if not date_str:
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")

    response = table.query(
        KeyConditionExpression="PK = :pk",
        ExpressionAttributeValues={":pk": f"STATS#{date_str}"}
    )
    return response["Items"]


def get_active_tasks():
    """List all currently active CAPTCHA tasks."""
    response = table.query(
        IndexName="GSI1",
        KeyConditionExpression="GSI1PK = :pk",
        ExpressionAttributeValues={":pk": "STATUS#polling"}
    )
    return response["Items"]

Aus get_daily_stats lässt sich die Erfolgsquote direkt berechnen: successful / total_solves. Die durchschnittliche Lösungszeit ergibt sich aus total_elapsed / successful. Weil beide Werte pro CAPTCHA-Typ vorliegen, sehen Sie sofort, ob etwa reCAPTCHA v2 und Cloudflare Turnstile unterschiedlich abschneiden.

JavaScript-Implementierung (AWS SDK v3)

Wer seine Lambda-Funktionen in Node.js schreibt, bildet dieselbe Logik mit dem AWS SDK v3 und dem Document Client ab. Die Schlüsselstruktur bleibt identisch, sodass Python- und Node-Worker in dieselbe Tabelle schreiben können.

const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { DynamoDBDocumentClient, PutCommand, QueryCommand, UpdateCommand } = require("@aws-sdk/lib-dynamodb");
const axios = require("axios");

const client = DynamoDBDocumentClient.from(new DynamoDBClient({}));
const TABLE = process.env.DYNAMODB_TABLE || "CaptchaSolves";
const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveAndTrack(sitekey, pageurl, type = "recaptcha_v2") {
  const now = new Date();
  const timestamp = now.toISOString();
  const ttl = Math.floor(now.getTime() / 1000) + 90 * 24 * 3600;

  const submit = await axios.post("https://ocr.captchaai.com/in.php", null, {
    params: { key: API_KEY, method: "userrecaptcha", googlekey: sitekey, pageurl, json: 1 },
  });

  if (submit.data.status !== 1) {
    await client.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `SITE#${sitekey}`, SK: `SOLVE#${timestamp}`, status: "error", error: submit.data.request, ttl },
    }));
    return { error: submit.data.request };
  }

  const captchaId = submit.data.request;
  let polls = 0;

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    polls++;
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
    });

    if (poll.data.status === 1) {
      const elapsed = Date.now() - now.getTime();
      await client.send(new PutCommand({
        TableName: TABLE,
        Item: {
          PK: `SOLVE#${captchaId}`, SK: "META", captcha_type: type,
          sitekey, pageurl, status: "solved", submitted_at: timestamp,
          solved_at: new Date().toISOString(), elapsed_ms: elapsed, polls, ttl,
        },
      }));
      return { solution: poll.data.request };
    }

    if (poll.data.request !== "CAPCHA_NOT_READY") {
      return { error: poll.data.request };
    }
  }
  return { error: "TIMEOUT" };
}

async function getSiteHistory(sitekey, limit = 50) {
  const result = await client.send(new QueryCommand({
    TableName: TABLE,
    KeyConditionExpression: "PK = :pk",
    ExpressionAttributeValues: { ":pk": `SITE#${sitekey}` },
    ScanIndexForward: false,
    Limit: limit,
  }));
  return result.Items;
}

Kostenoptimierung in DynamoDB

Die Datenmenge pro Solve ist klein, doch bei hohem Durchsatz summieren sich Schreib- und Lesevorgänge. Die folgenden Stellschrauben halten die Rechnung niedrig, ohne die Auswertbarkeit einzuschränken.

Strategie Wirkung
On-Demand-Abrechnung für schwankende Last nutzen Keine Überprovisionierung
TTL für die automatische Datensatzbereinigung aktivieren Senkt die Speicherkosten
In Abfragen nur die benötigten Attribute projizieren Weniger gelesene Kapazitätseinheiten
Schreibvorgänge mit BatchWriteItem bündeln Weniger API-Aufrufe
DynamoDB Streams für Auswertungen einsetzen Aggregation an Lambda auslagern

Beispiel aus der Praxis: Ein DACH-Datenteam betreibt seine Worker auf Hetzner-VPS und ruft CaptchaAI aus AWS Lambda auf. Pro Tag fallen rund 10.000 Solves an, aufgeteilt auf mehrere Projekte über das Attribut project. Mit On-Demand-Abrechnung und aktiviertem TTL bleibt die DynamoDB-Rechnung im niedrigen einstelligen Dollarbereich pro Monat – der Löwenanteil der Kosten entfällt ohnehin auf das CAPTCHA-Lösen selbst, nicht auf die Protokollierung.

Hinweis zum Datenschutz: pageurl und sitekey können Rückschlüsse auf die Ziel-Website zulassen. Wer in der EU arbeitet, sollte prüfen, welche Felder wirklich nötig sind, und die TTL bewusst knapp halten. Gespeicherte Daten, die niemand mehr auswertet, sind auch aus DSGVO-Sicht ein vermeidbares Risiko.

Fehlerbehebung

Problem Ursache Lösung
ProvisionedThroughputExceededException Zu viele Schreibvorgänge pro Sekunde Auf On-Demand-Abrechnung umstellen oder die WCU erhöhen
TTL-Elemente verschwinden nicht sofort DynamoDB löscht TTL-Elemente verzögert (bis zu ~48 Stunden) Nicht auf Echtzeit-Löschung verlassen; abgelaufene Elemente in Abfragen filtern
Hot Partition auf STATS#{date} Alle Worker schreiben in dieselbe Partition Zufälligen Suffix anhängen: STATS#{date}#shard{0-9}
Abfrage liefert zu viele Elemente Zu breiter Partitionsschlüssel SK-Bedingungen ergänzen, um die Ergebnisse einzugrenzen
Solve-Datensatz fehlt in der Tabelle Ausnahme vor dem DynamoDB-Write Write in einen try/finally-Block legen, damit er immer ausgeführt wird

Häufige Fragen (FAQ)

Wie lange sollte ich Lösungsdatensätze aufbewahren?

Im Beispiel läuft die TTL nach 90 Tagen ab. Für reine Betriebsauswertungen reichen oft schon 30 Tage. Legen Sie die Aufbewahrungsdauer bewusst fest, statt Daten unbegrenzt zu horten – DynamoDB entfernt abgelaufene Elemente dann von selbst, ohne zusätzlichen Bereinigungsjob.

Wie berechne ich die Erfolgsquote aus den gespeicherten Daten?

Aus der STATS#{date}-Partition: successful / total_solves ergibt die Erfolgsquote, total_elapsed / successful die durchschnittliche Lösungszeit. Weil die Werte pro CAPTCHA-Typ vorliegen, lassen sich reCAPTCHA v2, reCAPTCHA v3 und Turnstile getrennt bewerten.

Beeinflusst das Tracking meine CaptchaAI-Kosten?

Nein. CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung – von BASIC (15 $/Monat, 5 Threads) bis VIP-3 (7.500 $/Monat, 5.000 Threads), jeweils mit unbegrenzten Lösungen pro Thread. Die DynamoDB-Kosten für die Protokollierung sind davon vollständig getrennt und fallen bei AWS an.

Kann ich Solves aus vielen parallelen Lambda-Aufrufen sicher zählen?

Ja. Die atomare UpdateExpression mit if_not_exists in update_daily_stats sorgt dafür, dass gleichzeitige Zähler-Updates korrekt serialisiert werden. Sie brauchen keine eigene Sperre und riskieren keine verlorenen Schreibvorgänge.

Verwandte Leitfäden

Kommentare sind für diesen Artikel deaktiviert.