Calling CaptchaAI From a Dagster Asset

This guide wraps CaptchaAI in a Dagster ConfigurableResource and calls it from a daily-partitioned asset that loads a form, solves its reCAPTCHA v2 and submits it in one requests.Session. Dagster's own primitives then do the reliability work: RetryPolicy for transient failures, Failure(allow_retries=False) for permanent ones, a concurrency pool for throughput and an asset check for solve health. The constraint behind every choice is your plan's thread count. Each in-flight solve holds one thread, and a 30-day backfill will try to run more solves at once than the plan has threads unless something stops it.

The running example is a partner reporting portal that your company has a data agreement with. Its export form sits behind reCAPTCHA v2, and each day needs four regional exports. Use the pattern only on portals you own or are authorized to automate. Prefect 3 and Mage versions close the page.

How the pieces fit

1. Resource. CaptchaAIResource holds the key and turns CaptchaAI's responses into three exception classes. Assets declare it as a typed parameter, so tests can swap in a fake.
2. Asset. portal_records fetches the page, solves and posts the form in one session per region, then returns a MaterializeResult.
3. Pool. pool="captchaai" caps how many of those steps run at once across all runs.
4. Health. Metadata records solve counts and latency; a blocking asset check fails partitions with too many unsolvable regions.

The CaptchaAI calls and what each answer means

A solve is two HTTP calls to https://ocr.captchaai.com. The submit is a POST to /in.php with key, method=userrecaptcha, googlekey (the widget's data-sitekey), pageurl and json=1. It answers {"status":1,"request":"<task ID>"}. The poll is a GET to /res.php with key, action=get, id and json=1. It returns CAPCHA_NOT_READY (spelled without the T) until the token arrives in request. CaptchaAI's reCAPTCHA guidance is a first poll 15-20 seconds after submitting, then one every 5 seconds. The polling and timeout strategy guide covers the trade-offs for other types.

Two account actions matter for pipelines. action=threadsinfo returns {"threads":"5","working_threads":3}: your plan's thread cap (as a string) and how many are busy. action=getbalance returns the plan's thread count, not money. CaptchaAI bills per concurrent thread with unlimited solves, so threads are the budget a pipeline can exhaust. Each documented response maps to one exception and one Dagster behavior:

IP_BANNED counts as permanent: repeated bad-key requests cause it, and its 5-minute ban outlasts every retry delay here. The CaptchaAI error codes reference lists the full set with fixes.

The resource

# portal_pipeline/captchaai.py
import time

import dagster as dg
import requests

API = "https://ocr.captchaai.com"
TRANSIENT = {"ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR"}


class CaptchaPermanentError(Exception):
    """Bad key, bad parameters, IP_BANNED, no active plan: a retry cannot help."""


class CaptchaTransientError(Exception):
    """Server error, busy threads, network failure or poll timeout: retry later."""


class CaptchaUnsolvableError(Exception):
    """ERROR_CAPTCHA_UNSOLVABLE: stop polling this ID; a fresh task may work."""


def classify(code: str) -> Exception:
    if code == "ERROR_CAPTCHA_UNSOLVABLE":
        return CaptchaUnsolvableError(code)
    if code in TRANSIENT or not code.startswith(("ERROR_", "IP_BANNED")):
        return CaptchaTransientError(code)  # also HTML 5xx pages (HTTP_502 ...)
    return CaptchaPermanentError(code)


class CaptchaAIResource(dg.ConfigurableResource):
    api_key: str
    timeout_s: int = 120
    first_poll_s: int = 15
    poll_every_s: int = 5

    def _call(self, verb: str, path: str, params: dict) -> dict:
        params = {"key": self.api_key, "json": 1, **params}
        try:
            if verb == "POST":
                resp = requests.post(f"{API}/{path}", data=params, timeout=30)
            else:
                resp = requests.get(f"{API}/{path}", params=params, timeout=30)
        except requests.RequestException as exc:
            # "from None": the chained error would print the GET URL, key included
            raise CaptchaTransientError(f"network: {type(exc).__name__}") from None
        try:
            return resp.json()
        except ValueError:  # plain-text code despite json=1, or an HTML error page
            text = resp.text.strip()
            code = text if text.startswith(("ERROR_", "IP_BANNED")) else f"HTTP_{resp.status_code}"
            return {"status": 0, "request": code}

    def threads(self) -> dict:
        body = self._call("GET", "res.php", {"action": "threadsinfo"})
        if "threads" not in body:
            raise classify(str(body.get("request")))
        return {"threads": int(body["threads"]), "working_threads": int(body["working_threads"])}

    def submit(self, params: dict) -> str:
        body = self._call("POST", "in.php", params)
        if body.get("status") == 1:
            return str(body["request"])
        code = str(body.get("request"))
        if code == "ERROR_ZERO_BALANCE":  # no free thread, or no active plan
            info = self.threads()
            if info["working_threads"] < info["threads"]:
                raise CaptchaPermanentError("ERROR_ZERO_BALANCE with free threads: no active plan")
            raise CaptchaTransientError("ERROR_ZERO_BALANCE: all plan threads busy")
        raise classify(code)

    def poll(self, task_id: str, deadline: float) -> str:
        time.sleep(self.first_poll_s)
        while True:
            body = self._call("GET", "res.php", {"action": "get", "id": task_id})
            if body.get("status") == 1:
                return str(body["request"])
            code = str(body.get("request"))
            if code != "CAPCHA_NOT_READY" and not code.startswith("HTTP_"):
                raise classify(code)
            if time.monotonic() + self.poll_every_s > deadline:
                raise CaptchaTransientError(f"task {task_id} not ready after {self.timeout_s}s")
            time.sleep(self.poll_every_s)

    def solve(self, params: dict) -> tuple[str, str, float]:
        """Return (token, task_id, seconds). reCAPTCHA v2 answers arrive in `request`."""
        started = time.monotonic()
        task_id = self.submit(params)
        token = self.poll(task_id, deadline=started + self.timeout_s)
        return token, task_id, time.monotonic() - started

Three details in it are easy to get wrong:

• Response parsing. The API sometimes returns a plain-text error code even with json=1, and in rare cases an HTML 500 or 502 page. _call turns both into the usual {"status":0,"request":...} shape, and the poll loop treats an HTML page as "try again in 5 seconds", as CaptchaAI's API docs advise.
• ERROR_ZERO_BALANCE needs a diagnosis. It means "no free threads": either a saturated plan, worth a delayed retry, or no active plan, which no retry fixes. threadsinfo tells them apart.
• The deadline runs from submit. timeout_s bounds the whole solve. A task you stop polling can stay in flight on CaptchaAI's side, holding its thread, which is one reason to leave a spare thread.

The asset: load, solve and submit in one session

The portal logic calls no Dagster API, so the Prefect flow below reuses it unchanged:

# portal_pipeline/portal.py
import os
import re

import requests

from .captchaai import CaptchaPermanentError, CaptchaUnsolvableError

PORTAL = os.environ.get("PORTAL_BASE_URL", "https://reports.partner.example")
REGIONS = ["north", "south", "east", "west"]


def export_region(solver, day: str, region: str) -> tuple[list[dict], str, float]:
    # One Session per export: page load, solve and form POST share the same cookies.
    with requests.Session() as s:
        page = s.get(f"{PORTAL}/export", params={"day": day, "region": region}, timeout=30)
        page.raise_for_status()
        found = re.search(r'data-sitekey="([^"]+)"', page.text)
        if not found:
            raise CaptchaPermanentError("no reCAPTCHA sitekey on the export page")
        token, task_id, took = solver.solve(
            {"method": "userrecaptcha", "googlekey": found.group(1), "pageurl": page.url}
        )
        # Post at once: a reCAPTCHA token is valid for two minutes and verifies once.
        resp = s.post(
            f"{PORTAL}/export",
            data={"day": day, "region": region, "g-recaptcha-response": token},
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()["records"], task_id, took


def export_with_one_resubmit(solver, day: str, region: str) -> tuple[list[dict], str, float]:
    try:
        return export_region(solver, day, region)
    except CaptchaUnsolvableError:
        return export_region(solver, day, region)  # fresh page, fresh task ID

The token goes into the form in the same function that obtained it. Google documents that a reCAPTCHA response token is valid for two minutes and can be verified only once. Pre-solving tokens for a batch, or handing one to another step, produces tokens the portal rejects. The cookies the portal set on the page load also live in that one session. For more on submitting the token with requests, see solving reCAPTCHA v2 with Python Requests.

# portal_pipeline/assets.py
import concurrent.futures as cf
import json
import math
import os
from pathlib import Path

import dagster as dg

from .captchaai import CaptchaAIResource, CaptchaPermanentError, CaptchaUnsolvableError
from .portal import REGIONS, export_with_one_resubmit

WORKERS_PER_STEP = 2  # pool limit x WORKERS_PER_STEP must stay <= plan threads
daily = dg.DailyPartitionsDefinition(start_date="2026-09-01")


@dg.asset(
    partitions_def=daily,
    pool="captchaai",
    retry_policy=dg.RetryPolicy(
        max_retries=3, delay=10, backoff=dg.Backoff.EXPONENTIAL, jitter=dg.Jitter.PLUS_MINUS
    ),
)
def portal_records(context: dg.AssetExecutionContext, captchaai: CaptchaAIResource) -> dg.MaterializeResult:
    day = context.partition_key
    rows, durations, unsolvable = [], [], []
    with cf.ThreadPoolExecutor(max_workers=WORKERS_PER_STEP) as executor:
        futures = {executor.submit(export_with_one_resubmit, captchaai, day, r): r for r in REGIONS}
        for fut in cf.as_completed(futures):
            region = futures[fut]
            try:
                records, task_id, took = fut.result()
            except CaptchaUnsolvableError:
                unsolvable.append(region)
                context.log.warning(f"{day}/{region}: unsolvable twice, skipped")
                continue
            except CaptchaPermanentError as exc:
                executor.shutdown(cancel_futures=True)  # queued regions never start
                raise dg.Failure(description=f"{region}: {exc}", allow_retries=False) from exc
            except Exception:
                executor.shutdown(cancel_futures=True)  # the retry re-runs every region anyway
                raise
            context.log.info(f"{day}/{region}: captcha {task_id} solved in {took:.1f}s")
            rows.extend(records)
            durations.append(took)

    out = Path(os.environ.get("PORTAL_DATA_DIR", "data")) / f"portal_records_{day}.jsonl"
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text("".join(json.dumps(r) + "\n" for r in rows), encoding="utf-8")
    p95 = sorted(durations)[math.ceil(0.95 * len(durations)) - 1] if durations else 0.0
    return dg.MaterializeResult(
        metadata={
            "solves": len(durations),
            "unsolvable": len(unsolvable),
            "p95_solve_seconds": round(p95, 1),
            "rows": len(rows),
            "path": dg.MetadataValue.path(str(out)),
        }
    )


@dg.asset_check(asset=portal_records, blocking=True)
def unsolvable_ratio(context: dg.AssetCheckExecutionContext) -> dg.AssetCheckResult:
    partition = context.run.tags.get("dagster/partition")  # this run's partition
    records = context.instance.fetch_materializations(
        dg.AssetRecordsFilter(portal_records.key, asset_partitions=[partition] if partition else None),
        limit=1,
    ).records
    if not records:
        return dg.AssetCheckResult(passed=False, description="no materialization to check")
    md = records[0].asset_materialization.metadata
    solves, unsolvable = md["solves"].value, md["unsolvable"].value
    ratio = unsolvable / max(solves + unsolvable, 1)
    return dg.AssetCheckResult(passed=ratio <= 0.25, metadata={"unsolvable_ratio": round(ratio, 3)})

Each exception class reaches a different Dagster outcome:

• Permanent errors become dg.Failure(..., allow_retries=False). Dagster's API reference describes allow_retries as whether the failure should "respect the retry policy or bypass it and immediately fail". A revoked key fails the partition once instead of four times. Regions that have not started are cancelled; any already solving finish first.
• Transient errors propagate unchanged, so RetryPolicy re-runs the step. Regions that have not started are cancelled here too: the retry repeats every region, so starting them now would only spend threads twice. With Backoff.EXPONENTIAL, Dagster waits ((2^n) - 1) × delay before retry n: 10, 30 and 70 seconds here. Jitter.PLUS_MINUS moves each wait by up to 10 seconds, so partitions that failed together do not come back together.
• Unsolvable twice is recorded, not raised. Whether a partition missing one region is acceptable is a data-quality decision, and that belongs in the asset check.

Sizing the pool to your plan threads

The pool argument arrived in Dagster 1.9.12 and replaced the older tag-based concurrency keys. With the default op granularity, a pool limit caps how many steps assigned to that pool execute at once across every run in the deployment. See the concurrency pools documentation for the full model.

A pool slot is a step, not a solve. Each portal_records step runs up to WORKERS_PER_STEP solves at once, so the rule is pool limit × workers per step ≤ plan threads. On BASIC ($15/month, 5 threads), a limit of 2 uses at most 4 threads and leaves one for a manual rerun or a slow timed-out task. On STANDARD ($30/month, 15 threads), a limit of 7 uses 14. Pool limits are stored in the Dagster instance, so DAGSTER_HOME must point at the same directory for the CLI, dagster dev and your daemon:

pip install dagster dagster-webserver requests responses pytest
export DAGSTER_HOME="$HOME/dagster_home" && mkdir -p "$DAGSTER_HOME"
export CAPTCHAAI_API_KEY="YOUR_API_KEY"   # read by dg.EnvVar at run time
export PORTAL_BASE_URL="https://reports.partner.example"

# BASIC plan: 5 threads, 2 solves per step -> at most 2 steps in the pool
dagster instance concurrency set captchaai 2
dagster instance concurrency get captchaai

dagster asset materialize -m portal_pipeline.definitions --select portal_records --partition 2026-09-01
dagster dev -m portal_pipeline.definitions

The limit can also be added in the UI's deployment concurrency settings with Add pool limit. By default, a backfill launches one run per partition, so a 30-day backfill brings 30 runs that all want the same two slots. A run tag limit on dagster/backfill bounds how many runs each backfill has in progress at once, which leaves run capacity for your other jobs. Put this in $DAGSTER_HOME/dagster.yaml:

concurrency:
  runs:
    max_concurrent_runs: 10
    tag_concurrency_limits:
      - key: "dagster/backfill"
        value:
          applyLimitPerUniqueValue: true
        limit: 3
run_monitoring:
  enabled: true
  free_slots_after_run_end_seconds: 300

The run_monitoring block matters more than it looks. Dagster's docs warn that pool slots held by a cancelled or failed run are not freed by default, and one stale slot can block the pool. free_slots_after_run_end_seconds releases them.

On Dagster versions before 1.9.12, tag the asset with op_tags={"dagster/concurrency_key": "captchaai"} and set the limit with the same dagster instance concurrency set command. Run-queue limits went under a top-level run_queue: key instead of concurrency: runs:, which only exists from that release on.

The pool only sees Dagster. If an Airflow DAG, a Celery worker fleet or an ad-hoc script uses the same key, subtract their share, or poll threads() and alert when working_threads sits at the cap. The thread-limit guide covers when that means tuning and when it means a bigger plan.

Why blocking while polling is acceptable

A Dagster step does not hand its wait back to the orchestrator. While a solve is in progress, the worker thread sleeps, and the step keeps its process and its pool slot. At this scale that is cheap: CaptchaAI typically solves reCAPTCHA v2 in under 60 seconds, so four regions at two workers finish in about two solve cycles.

Budget timeouts for the worst case, though. One solve can take about 150 seconds: the 120-second poll deadline plus HTTP request timeouts. An unsolvable region is solved twice, and four regions need two rounds of workers, so solve time alone can bring one attempt close to 10 minutes. Four attempts approach 40 minutes, plus the retry delays and the page loads. If you set the dagster/max_runtime run tag, keep it above that plus any wait for a pool slot, or run monitoring fails healthy runs that are only waiting on CAPTCHAs.

When held workers become the bottleneck, at thousands of solves a day, move the wait out of the worker. The deferrable Airflow operator releases its worker slot between polls. Durable workflow engines such as Temporal sleep on timers that survive restarts.

Solve health: metadata and a blocking check

The metadata records solves, unsolvable, p95_solve_seconds and rows on every materialization. Dagster treats numeric metadata as a time series and plots it on the asset's page, so a p95 creeping from 30 to 90 seconds is visible before any partition fails.

The unsolvable_ratio check fails when more than a quarter of the regions came back unsolvable. With blocking=True and the default ERROR severity, a failing check stops downstream assets in the same run from materializing. The check is not partitioned, because Dagster still flags a partitions_def on an asset check as a preview feature. Instead it reads the materialization for its own run's dagster/partition tag. Reading the latest materialization would be a race: with two pool slots, another partition can finish between the asset and its check.

Logs carry the CaptchaAI task ID for every solve and never the key. The key comes from dg.EnvVar, which Dagster resolves at run time and does not display in the UI. Tracebacks are the leak to watch: a requests connection error on a GET quotes the full URL, key= query parameter included, and Dagster shows the whole exception chain on the failed step. That is why _call raises network errors from None.

Wiring it together

# portal_pipeline/definitions.py
import dagster as dg

from .assets import portal_records, unsolvable_ratio
from .captchaai import CaptchaAIResource

portal_daily = dg.define_asset_job("portal_daily", selection=[portal_records])

defs = dg.Definitions(
    assets=[portal_records],
    asset_checks=[unsolvable_ratio],
    schedules=[dg.build_schedule_from_partitioned_job(portal_daily, hour_of_day=3)],
    resources={"captchaai": CaptchaAIResource(api_key=dg.EnvVar("CAPTCHAAI_API_KEY"))},
)

The job selects the asset, and Dagster includes checks that target it, so every scheduled partition runs the check too. dagster definitions validate -m portal_pipeline.definitions catches wiring mistakes before a deploy. Recent releases mark it superseded by dg check defs, which does the same for projects built with the dg CLI.

Testing without spending threads

Because the asset asks for a CaptchaAIResource, a subclass that overrides solve() is a drop-in fake. responses stubs the portal and, for the classification test, CaptchaAI itself:

# tests/test_portal.py
import dagster as dg
import pytest
import responses

from portal_pipeline.assets import portal_records, unsolvable_ratio
from portal_pipeline.captchaai import CaptchaAIResource, CaptchaPermanentError, CaptchaTransientError
from portal_pipeline.portal import PORTAL


class FakeCaptchaAI(CaptchaAIResource):
    def solve(self, params: dict) -> tuple[str, str, float]:
        return "fake-token", "1234567890", 12.0


@responses.activate
def test_partition_materializes(tmp_path, monkeypatch):
    monkeypatch.setenv("PORTAL_DATA_DIR", str(tmp_path))
    responses.get(f"{PORTAL}/export", body='<div class="g-recaptcha" data-sitekey="6Lc_test"></div>')
    responses.post(f"{PORTAL}/export", json={"records": [{"id": 1}]})
    result = dg.materialize(
        [portal_records, unsolvable_ratio],
        resources={"captchaai": FakeCaptchaAI(api_key="test")},
        partition_key="2026-09-01",
    )
    assert result.success
    md = result.asset_materializations_for_node("portal_records")[0].metadata
    assert md["solves"].value == 4 and md["unsolvable"].value == 0


@responses.activate
def test_error_classification():
    api = CaptchaAIResource(api_key="test", first_poll_s=0, poll_every_s=0)
    params = {"method": "userrecaptcha", "googlekey": "k", "pageurl": "https://portal.example"}
    responses.post("https://ocr.captchaai.com/in.php", json={"status": 0, "request": "ERROR_WRONG_USER_KEY"})
    with pytest.raises(CaptchaPermanentError):
        api.submit(params)
    responses.replace(responses.POST, "https://ocr.captchaai.com/in.php",
                      json={"status": 0, "request": "ERROR_ZERO_BALANCE"})
    responses.get("https://ocr.captchaai.com/res.php", json={"threads": "5", "working_threads": 5})
    with pytest.raises(CaptchaTransientError):  # all five threads busy: retry later
        api.submit(params)

For shared fixtures across a larger suite, see pytest fixtures for CaptchaAI.

The same pattern in Prefect 3

Prefect does not inject resources into tasks, so the task builds the client itself and reads the key from a Secret block. A global concurrency limit plays the pool's role. Each task solves one region at a time, so here one slot is one thread. A limit of 4 on BASIC keeps the same spare thread as the Dagster pool. Create the limit before the first run: without a limit named captchaai, Prefect logs a warning and runs unthrottled (strict=True makes it fail instead).

pip install prefect requests
prefect gcl create captchaai --limit 4   # BASIC: 5 threads, one spare
python -c 'import os; from prefect.blocks.system import Secret; Secret(value=os.environ["CAPTCHAAI_API_KEY"]).save("captchaai-api-key")'

# prefect_flow.py
from prefect import flow, task
from prefect.blocks.system import Secret
from prefect.concurrency.sync import concurrency
from prefect.tasks import exponential_backoff

from portal_pipeline.captchaai import CaptchaAIResource, CaptchaPermanentError
from portal_pipeline.portal import REGIONS, export_with_one_resubmit


def retry_unless_permanent(task, task_run, state) -> bool:
    try:
        state.result()
    except CaptchaPermanentError:
        return False  # bad key or parameters: fail now
    except Exception:
        return True
    return True


@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(backoff_factor=10),
    retry_jitter_factor=0.5,
    retry_condition_fn=retry_unless_permanent,
)
def export_region_task(day: str, region: str) -> list[dict]:
    solver = CaptchaAIResource(api_key=Secret.load("captchaai-api-key").get())
    with concurrency("captchaai", occupy=1):  # one slot per in-flight solve
        records, task_id, took = export_with_one_resubmit(solver, day, region)
    print(f"{day}/{region}: captcha {task_id} solved in {took:.1f}s")
    return records


@flow(log_prints=True)
def portal_records_flow(day: str) -> int:
    futures = [export_region_task.submit(day, r) for r in REGIONS]
    return sum(len(f.result()) for f in futures)

retry_condition_fn is Prefect's counterpart to allow_retries=False: it receives the failed state and returns False for errors a retry cannot fix. The resource class works outside Dagster, but a Prefect-only environment should not need dagster installed. In captchaai.py, swap import dagster as dg for from dataclasses import dataclass and decorate a plain class CaptchaAIResource: with @dataclass; the four fields and every method stay the same.

The same pattern in Mage

A Mage data loader block reads the key with get_secret_value from a secret created on Mage's Secrets page. The block keeps its own compact client with the same permanent/transient split and a bounded poll loop:

import time
from datetime import datetime

import requests

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test
from mage_ai.data_preparation.shared.secrets import get_secret_value

API = 'https://ocr.captchaai.com'
PORTAL = 'https://reports.partner.example'
TRANSIENT = {'ERROR_SERVER_ERROR', 'ERROR_INTERNAL_SERVER_ERROR', 'ERROR_ZERO_BALANCE',
             'ERROR_CAPTCHA_UNSOLVABLE'}  # a block retry starts fresh pages and tasks


class CaptchaPermanentError(Exception):
    pass


class CaptchaTransientError(Exception):
    pass


def answer_or_raise(body: dict) -> str:
    code = str(body.get('request'))
    if body.get('status') == 1:
        return code
    if code == 'CAPCHA_NOT_READY':
        return ''
    raise (CaptchaTransientError if code in TRANSIENT else CaptchaPermanentError)(code)


def solve_recaptcha_v2(key: str, sitekey: str, pageurl: str, timeout_s: int = 120) -> str:
    deadline = time.monotonic() + timeout_s
    task_id = answer_or_raise(requests.post(f'{API}/in.php', data={
        'key': key, 'method': 'userrecaptcha', 'googlekey': sitekey,
        'pageurl': pageurl, 'json': 1}, timeout=30).json())
    time.sleep(15)
    while time.monotonic() < deadline:  # bounded: never poll one task forever
        # res.php takes POST too; a form body keeps the key out of error URLs
        token = answer_or_raise(requests.post(f'{API}/res.php', data={
            'key': key, 'action': 'get', 'id': task_id, 'json': 1}, timeout=30).json())
        if token:
            return token
        time.sleep(5)
    raise CaptchaTransientError(f'task {task_id} not ready after {timeout_s}s')


@data_loader
def load_portal_records(*args, **kwargs):
    key = get_secret_value('CAPTCHAAI_API_KEY')
    day = (kwargs.get('execution_date') or datetime.now()).date().isoformat()
    rows = []
    for region in ('north', 'south', 'east', 'west'):
        with requests.Session() as s:  # page, solve and POST share one session
            page = s.get(f'{PORTAL}/export', params={'day': day, 'region': region}, timeout=30)
            page.raise_for_status()
            sitekey = page.text.split('data-sitekey="')[1].split('"')[0]
            token = solve_recaptcha_v2(key, sitekey, page.url)
            resp = s.post(f'{PORTAL}/export', timeout=60, data={
                'day': day, 'region': region, 'g-recaptcha-response': token})
            resp.raise_for_status()
            rows.extend(resp.json()['records'])
    return rows


@test
def test_output(output, *args) -> None:
    assert output, 'The export returned no records'

Retries and concurrency go in the pipeline's metadata.yaml. Add retry_config to this block's entry under blocks:, and a concurrency_config at pipeline level. The block solves one region at a time, so each pipeline run holds at most one thread. Four concurrent runs across all triggers fit BASIC's 5 threads with one spare:

concurrency_config:
  pipeline_run_limit_all_triggers: 4
blocks:
- uuid: load_portal_records  # keep the entry's existing keys; add retry_config
  type: data_loader
  retry_config:
    retries: 2
    delay: 10
    max_delay: 120
    exponential_backoff: true

Mage's block retry settings apply to every failed block run and document no per-exception opt-out. Busy threads and unsolvable tasks count as transient in this block, because a retry reloads the pages and submits fresh tasks. Keep retries low and let the exception name tell whoever is on call that CaptchaPermanentError: ERROR_WRONG_USER_KEY is not worth rerunning.

Troubleshooting

A backfill makes every solve slow, or submits fail with ERROR_ZERO_BALANCE

The pool limit is missing, set on a different DAGSTER_HOME, or sized to threads instead of threads ÷ workers per step. Saturation does not always surface as an error; it can show up only as longer CAPCHA_NOT_READY waits and a rising p95_solve_seconds. Run dagster instance concurrency get captchaai with the daemon's DAGSTER_HOME, and compare working_threads from threads() with your plan. Other consumers of the same key count too.

Retries keep burning on an invalid key

Something wraps the permanent error before it reaches the asset, so Dagster sees a plain exception and applies RetryPolicy. Check that CaptchaPermanentError reaches the except that raises dg.Failure(..., allow_retries=False). Stop quickly: repeated bad-key requests lead to IP_BANNED for 5 minutes.

The portal rejects a token CaptchaAI returned

The token was posted from a different session than the page load, used twice, or held longer than two minutes. Keep the fetch, solve and POST inside export_region. If the widget sits in an iframe on another domain and submits fail with ERROR_BAD_TOKEN_OR_PAGEURL, send the iframe's URL as pageurl.

Re-materializing a partition solves every region again

Each materialization fetches fresh pages and needs fresh tokens, so every rerun costs solves and thread time. Backfill only missing or failed partitions, not the whole range. Also, RetryPolicy re-runs the whole step, and regions that succeeded are solved again, because rows are written only at the end.

FAQ

Should the solve be its own asset?

No. Asset outputs are persisted and loaded by the next step, possibly in another process minutes later. A reCAPTCHA token lasts two minutes and belongs with the session that loaded the form, so fetch, solve and submit belong in one step.

Can I use the official captchaai Python SDK inside the resource?

Yes. pip install captchaai (Python 3.10 or newer) gives CaptchaAI(api_key), which validates the key with a threadsinfo call at construction, and solver.recaptcha_v2(sitekey, url). Pass max_retries=0 so the SDK does not retry submits on its own and Dagster's RetryPolicy stays in charge. Then map the SDK's InvalidKeyError, ValidationError and NoThreadsError to dg.Failure(allow_retries=False), and UnsolvableError to the unsolvable path. ThreadLimitError, NetworkError and the SDK's own TimeoutError can propagate to the retry policy. APIError is the awkward one: it covers IP_BANNED as well as server errors, so a retry helps only some of the time.

Where to start

A daily four-region export fits BASIC ($15/month, 5 threads). Set the pool limit to 2, materialize one partition, and read p95_solve_seconds before you launch the backfill.