{"id":25445,"date":"2026-08-29T09:58:15","date_gmt":"2026-08-29T09:58:15","guid":{"rendered":"https:\/\/capskip.com\/?p=25445"},"modified":"2026-08-29T09:58:15","modified_gmt":"2026-08-29T09:58:15","slug":"airflow-captcha-dag","status":"publish","type":"post","link":"https:\/\/capskip.com\/zh\/airflow-captcha-dag\/","title":{"rendered":"\u5982\u4f55\u5728 Apache Airflow DAG \u4e2d\u8bc6\u522b\u9a8c\u8bc1\u7801\uff08Python\uff09"},"content":{"rendered":"<p>An Airflow captcha step is an ordinary Python task. You call the solver, you get a token back, you use it in the same task. There is no operator to install and no plugin to write. What actually trips people up is location: your workers run wherever the scheduler put them, and a solver bound to your laptop&#8217;s loopback address is not reachable from a container on another host. Get that part right first, then the DAG is fifteen lines.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">What you need<\/h2>\n<ul>\n<li>Airflow 2.x or 3.x, with the CapSkip SDK installed in the same image or virtualenv your workers use.<\/li>\n<li>A target that returns a challenge. A page behind a widget, not a test key that always passes.<\/li>\n<li>CapSkip running in Server mode on a machine your workers can reach, or in Local mode if the worker and the solver are the same box. Both are described under <a href=\"https:\/\/capskip.com\/setup-guide\/#connection-settings\">connection settings<\/a>.<\/li>\n<\/ul>\n<div data-no-translation>\n<pre data-enlighter-language=\"bash\" class=\"EnlighterJSRAW\"># Install into the worker environment, not just the scheduler.\r\npip install capskip<\/pre>\n<\/div>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Where the solver has to live<\/h2>\n<p>This is the whole problem, so it goes first. A task runs inside a worker process, and on most real deployments that worker is a container on a different host from anything you administer by hand. The loopback address inside that container is the container, so pointing the SDK at it finds nothing.<\/p>\n<table>\n<thead>\n<tr>\n<th>Mode<\/th>\n<th>Listens on<\/th>\n<th>Use it when<\/th>\n<\/tr>\n<\/thead>\n<tbody>\n<tr>\n<td>Local<\/td>\n<td>127.0.0.1, that device only<\/td>\n<td>A single worker on the same machine as the solver<\/td>\n<\/tr>\n<tr>\n<td>Server<\/td>\n<td>Your network address or public IP<\/td>\n<td>Containers, a worker fleet, a VPS or a managed Airflow<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<p>Server mode is the answer for nearly every Airflow deployment. You switch the listen address in the app, point the SDK host at that machine, and every worker in the fleet shares one solver. A static public IP is recommended when the callers sit outside your own network. This does not change what the product is: it is still your hardware and still unmetered, so moving it off the loopback address moves where it runs and nothing else. CapSkip is a Windows application, so in practice that is one Windows box the fleet calls into.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Step 1: configure the host once<\/h2>\n<p>Do not hardcode the address in the DAG file. The SDK reads CAPSKIP_HOST, CAPSKIP_PORT and CAPSKIP_API_KEY from the environment, which is the cleanest fit for Airflow because you already have a way to set worker environment variables. An Airflow Variable works too if you would rather keep it in the metadata database.<\/p>\n<div data-no-translation>\n<pre data-enlighter-language=\"python\" class=\"EnlighterJSRAW\"># pip install capskip\r\nimport os\r\nfrom capskip import CapSkip\r\n\r\n\r\ndef get_solver():\r\n    &quot;&quot;&quot;One place that knows where the solver lives.&quot;&quot;&quot;\r\n    return CapSkip(\r\n        host=os.environ.get(&quot;CAPSKIP_HOST&quot;, &quot;127.0.0.1&quot;),\r\n        port=int(os.environ.get(&quot;CAPSKIP_PORT&quot;, &quot;8080&quot;)),\r\n        recaptchaTimeout=300,\r\n    )<\/pre>\n<\/div>\n<p>Build the client inside the task, not at module scope. Airflow parses every DAG file on a loop, and anything created at import time is created on every parse, in the scheduler as well as the worker.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Step 2: write the solve as one task<\/h2>\n<p>The TaskFlow decorators are the shortest way in. Airflow 3 imports them from the SDK module, Airflow 2 from the decorators module, and the body is identical either way.<\/p>\n<div data-no-translation>\n<pre data-enlighter-language=\"python\" class=\"EnlighterJSRAW\"># Airflow 3.x. On 2.x use: from airflow.decorators import dag, task\r\nfrom airflow.sdk import dag, task\r\n\r\nPAGE = &quot;https:\/\/example.com\/page-with-recaptcha&quot;\r\nSITEKEY = &quot;YOUR_SITEKEY&quot;\r\n\r\n\r\n@task(retries=2)\r\ndef fetch_protected_page():\r\n    solver = get_solver()\r\n\r\n    # Solve and use in the same task. The token is short lived.\r\n    token = solver.recaptcha(sitekey=SITEKEY, url=PAGE)[&quot;code&quot;]\r\n\r\n    return post_form(PAGE, token)<\/pre>\n<\/div>\n<p>That is the entire integration. The solver call is synchronous, it polls for you with a backoff that starts at a quarter of a second, and it returns a dictionary whose code field holds the token.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Step 3: do not pass the token between tasks<\/h2>\n<p>This is the mistake worth naming, because Airflow makes it feel natural. TaskFlow returns become XComs, so a solve task returning a token and a downstream task consuming it looks like good design. It is a race. A reCAPTCHA token stays valid for roughly two minutes, and the gap between two Airflow tasks is a scheduler decision you do not control. Add a queue delay, a pool that is full, or a worker restart, and the token expires in transit.<\/p>\n<p>The failure is intermittent, which is the worst kind: the DAG works in testing and fails a few percent of the time in production, with a rejected form and no error from the solver. <a href=\"https:\/\/capskip.com\/recaptcha-token-expiration\/\">The post on how long a reCAPTCHA token stays valid<\/a> covers the timings. In a DAG the rule reduces to one line: solve and submit inside the same task, and let the retry re-solve.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">The full DAG<\/h2>\n<div data-no-translation>\n<pre data-enlighter-language=\"python\" class=\"EnlighterJSRAW\"># pip install capskip\r\nimport os\r\nfrom datetime import datetime, timedelta\r\n\r\nimport requests\r\nfrom airflow.sdk import dag, task\r\nfrom capskip import CapSkip\r\n\r\nPAGE = &quot;https:\/\/example.com\/page-with-recaptcha&quot;\r\nSITEKEY = &quot;YOUR_SITEKEY&quot;\r\n\r\n\r\n@dag(\r\n    schedule=&quot;@hourly&quot;,\r\n    start_date=datetime(2026, 1, 1),\r\n    catchup=False,\r\n    tags=[&quot;scraping&quot;],\r\n)\r\ndef protected_source():\r\n\r\n    @task(retries=2, retry_delay=timedelta(minutes=2), pool=&quot;captcha&quot;)\r\n    def scrape():\r\n        solver = CapSkip(\r\n            host=os.environ.get(&quot;CAPSKIP_HOST&quot;, &quot;127.0.0.1&quot;),\r\n            port=int(os.environ.get(&quot;CAPSKIP_PORT&quot;, &quot;8080&quot;)),\r\n        )\r\n        token = solver.recaptcha(sitekey=SITEKEY, url=PAGE)[&quot;code&quot;]\r\n\r\n        # Same task, so the token is seconds old when it is used.\r\n        r = requests.post(\r\n            PAGE,\r\n            data={&quot;g-recaptcha-response&quot;: token},\r\n            timeout=60,\r\n        )\r\n        r.raise_for_status()\r\n        return len(r.text)\r\n\r\n    scrape()\r\n\r\n\r\nprotected_source()<\/pre>\n<\/div>\n<p>The pool argument is doing real work there. A pool caps how many task instances run at once across the whole deployment, so a backfill of two hundred runs does not open two hundred simultaneous solves against one machine. Create a pool named captcha in the UI, give it the number of parallel solves you want, and every task that names it queues behind that limit.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Retries that help rather than hurt<\/h2>\n<p>Set retries on the task and let the whole solve and submit repeat. Because the token is fetched inside the task body, a retry gets a fresh one automatically, which is the behaviour you want and the reason the two steps stay together.<\/p>\n<p>Give the retry a delay. An immediate retry against a site that just challenged you tends to be challenged again, and a couple of minutes costs nothing in a scheduled pipeline. Two retries is usually enough: a third failure is normally a wrong sitekey or a solver that is not reachable, and neither of those is fixed by waiting.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">Common errors<\/h2>\n<table>\n<thead>\n<tr>\n<th>What you see<\/th>\n<th>Cause<\/th>\n<th>Fix<\/th>\n<\/tr>\n<\/thead>\n<tbody>\n<tr>\n<td>NetworkException on every task<\/td>\n<td>The worker cannot reach the solver<\/td>\n<td>Switch to Server mode and set CAPSKIP_HOST on the worker<\/td>\n<\/tr>\n<tr>\n<td>Works on the scheduler, fails on the worker<\/td>\n<td>The SDK is missing from the worker image<\/td>\n<td>Install it where tasks run, not only where DAGs are parsed<\/td>\n<\/tr>\n<tr>\n<td>TimeoutException under load<\/td>\n<td>More concurrent solves than the machine handles<\/td>\n<td>Put the task in a pool and cap the slot count<\/td>\n<\/tr>\n<tr>\n<td>Form rejects a token that solved fine<\/td>\n<td>The token aged between two tasks<\/td>\n<td>Move the submit into the solving task<\/td>\n<\/tr>\n<tr>\n<td>ERROR_PAGEURL<\/td>\n<td>A relative URL reached the API<\/td>\n<td>Send the absolute URL, scheme included<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n<p>The full list of codes and what triggers them is in <a href=\"https:\/\/capskip.com\/api-docs\/\">the CapSkip API documentation<\/a>.<\/p>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">FAQ<\/h2>\n<details style=\"border:1px solid #e2e5ee;border-radius:10px;padding:14px 18px;margin:0 0 12px;\">\n<summary style=\"cursor:pointer;\">\n<h3 style=\"font-size:1.15rem;line-height:1.4;display:inline;margin:0;\">My workers run in containers. Can they reach the solver?<\/h3>\n<\/summary>\n<p style=\"margin:12px 0 0;\">Yes, in Server mode. The solver listens on a network address instead of the loopback address, and the workers call it over the API like any other internal service. Set the host and port as worker environment variables so the DAG file has no address in it. Nothing about the code changes between a container and a laptop.<\/p>\n<\/details>\n<details style=\"border:1px solid #e2e5ee;border-radius:10px;padding:14px 18px;margin:0 0 12px;\">\n<summary style=\"cursor:pointer;\">\n<h3 style=\"font-size:1.15rem;line-height:1.4;display:inline;margin:0;\">What about managed Airflow that I do not administer?<\/h3>\n<\/summary>\n<p style=\"margin:12px 0 0;\">Same answer, with one extra requirement. A hosted scheduler runs outside your network, so the solver needs an address it can route to, and a static public IP is what makes that stable. Turn on key validation and give each environment its own key, so a leaked value can be revoked without touching the others.<\/p>\n<\/details>\n<details style=\"border:1px solid #e2e5ee;border-radius:10px;padding:14px 18px;margin:0 0 12px;\">\n<summary style=\"cursor:pointer;\">\n<h3 style=\"font-size:1.15rem;line-height:1.4;display:inline;margin:0;\">Should the solve be its own task for observability?<\/h3>\n<\/summary>\n<p style=\"margin:12px 0 0;\">It is tempting and it costs you reliability. A separate task means the token travels as an XCom and ages while the scheduler decides what runs next, which is exactly how tokens expire in transit. Keep them together and get your observability from logs and task duration instead, which show the same thing without the race.<\/p>\n<\/details>\n<details style=\"border:1px solid #e2e5ee;border-radius:10px;padding:14px 18px;margin:0 0 12px;\">\n<summary style=\"cursor:pointer;\">\n<h3 style=\"font-size:1.15rem;line-height:1.4;display:inline;margin:0;\">Can several DAGs share one solver instance?<\/h3>\n<\/summary>\n<p style=\"margin:12px 0 0;\">Yes, and that is the usual setup. One instance in Server mode serves every worker that can reach it, and there is no per solve cost to divide up. Use an Airflow pool to bound total concurrency across DAGs, because the limit you care about is how many solves run at once, not how many pipelines happen to want one.<\/p>\n<\/details>\n<h2 style=\"font-size:1.6rem;line-height:1.35;\">The short version<\/h2>\n<p>Put the solver on a machine the workers can reach, set the host in the environment, and keep the solve and the submit in one task with a couple of retries behind a pool. Everything else is a normal DAG. For the crawl side of this see <a href=\"https:\/\/capskip.com\/captcha-solver-for-web-scraping\/\">the CAPTCHA solver for web scraping page<\/a>, and for the client surface see <a href=\"https:\/\/capskip.com\/python-captcha-solver\/\">the Python CAPTCHA solver page<\/a>. Scheduled work is where the pricing model shows up. An hourly DAG that solves on every run is a real bill from a metered vendor, and that is the difference here: <a href=\"https:\/\/capskip.com\/\">captcha solver<\/a> running on hardware you already own costs the same whether the DAG fires once a day or once a minute.<\/p>\n","protected":false},"excerpt":{"rendered":"<p>\u5728 Airflow \u91cc\u8bc6\u522b\u9a8c\u8bc1\u7801\u662f\u4e00\u4e2a\u666e\u901a\u7684 Python \u4efb\u52a1\u3002\u771f\u6b63\u628a\u4eba\u7eca\u5012\u7684\u5730\u65b9\uff0c\u662f\u8981\u4ece\u4e00\u4e2a\u4e0d\u662f\u4f60\u81ea\u5df1\u673a\u5668\u7684 worker \u4e0a\u8fde\u5230\u8bc6\u522b\u5de5\u5177\u3002<\/p>","protected":false},"author":1,"featured_media":25444,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"rank_math_title":"Airflow CAPTCHA: Solve It Inside a DAG Task | CapSkip","rank_math_description":"An Airflow captcha step is a normal Python task. The hard part is where the solver lives, because workers are not your laptop. Full TaskFlow DAG inside.","rank_math_focus_keyword":"airflow captcha","footnotes":""},"categories":[70],"tags":[],"class_list":["post-25445","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-captcha"],"_links":{"self":[{"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/posts\/25445","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/comments?post=25445"}],"version-history":[{"count":1,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/posts\/25445\/revisions"}],"predecessor-version":[{"id":25451,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/posts\/25445\/revisions\/25451"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/media\/25444"}],"wp:attachment":[{"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/media?parent=25445"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/categories?post=25445"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/capskip.com\/zh\/wp-json\/wp\/v2\/tags?post=25445"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}