This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 9491792d270 Fail fast on permanent Anthropic batch poll errors (#70177)
9491792d270 is described below
commit 9491792d270fb3fadedfe7227d5b43fff9fb1639
Author: suyua9 <[email protected]>
AuthorDate: Sun Nov 8 10:47:32 2026 +0800
Fail fast on permanent Anthropic batch poll errors (#70177)
---
.../airflow/providers/anthropic/hooks/anthropic.py | 21 +++++++++-
.../airflow/providers/anthropic/triggers/batch.py | 7 +++-
.../tests/unit/anthropic/hooks/test_anthropic.py | 46 +++++++++++++++++++++-
.../tests/unit/anthropic/triggers/test_batch.py | 28 +++++++++++++
4 files changed, 99 insertions(+), 3 deletions(-)
diff --git
a/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py
b/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py
index 0fe6a1b07f8..7a27776b23b 100644
--- a/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py
+++ b/providers/anthropic/src/airflow/providers/anthropic/hooks/anthropic.py
@@ -32,6 +32,7 @@ from anthropic import (
AnthropicBedrock,
AnthropicFoundry,
AnthropicVertex,
+ APIStatusError,
AsyncAnthropic,
AsyncAnthropicAWS,
AsyncAnthropicBedrock,
@@ -117,6 +118,23 @@ class _ClientFactories(Generic[_ClientT]):
MAX_CONSECUTIVE_POLL_FAILURES = 5
+def _is_permanent_poll_error(error: BaseException) -> bool:
+ """
+ Return whether a failed poll is an API error that retrying will not fix.
+
+ Follows the SDK's own retry rule: an ``x-should-retry`` response header
decides when
+ present, otherwise 408, 409, 429 and 5xx are retryable and any other
status is not.
+ Errors without an HTTP status (network failures, timeouts) count as
transient.
+ """
+ if not isinstance(error, APIStatusError):
+ return False
+ should_retry = error.response.headers.get("x-should-retry")
+ if should_retry in ("true", "false"):
+ return should_retry == "false"
+ status = error.status_code
+ return status not in (408, 409, 429) and status < 500
+
+
class BatchStatus(str, Enum):
"""Top-level ``processing_status`` of an Anthropic Message Batch."""
@@ -676,7 +694,8 @@ class AnthropicHook(BaseHook):
# results remain recoverable for 29 days.
consecutive_failures += 1
if (
- consecutive_failures >= MAX_CONSECUTIVE_POLL_FAILURES
+ _is_permanent_poll_error(e)
+ or consecutive_failures >= MAX_CONSECUTIVE_POLL_FAILURES
or time.monotonic() - start > timeout
):
raise
diff --git
a/providers/anthropic/src/airflow/providers/anthropic/triggers/batch.py
b/providers/anthropic/src/airflow/providers/anthropic/triggers/batch.py
index cdb6973b7d9..65fb738baf0 100644
--- a/providers/anthropic/src/airflow/providers/anthropic/triggers/batch.py
+++ b/providers/anthropic/src/airflow/providers/anthropic/triggers/batch.py
@@ -25,6 +25,7 @@ from airflow.providers.anthropic.hooks.anthropic import (
MAX_CONSECUTIVE_POLL_FAILURES,
AnthropicHook,
BatchStatus,
+ _is_permanent_poll_error,
)
from airflow.triggers.base import BaseTrigger, TriggerEvent
@@ -88,7 +89,11 @@ class AnthropicBatchTrigger(BaseTrigger):
except Exception as e:
# Tolerate transient polling errors rather than failing a (up
to 24h) wait.
consecutive_failures += 1
- if consecutive_failures >= MAX_CONSECUTIVE_POLL_FAILURES or
time.time() > self.end_time:
+ if (
+ _is_permanent_poll_error(e)
+ or consecutive_failures >= MAX_CONSECUTIVE_POLL_FAILURES
+ or time.time() > self.end_time
+ ):
yield TriggerEvent({"status": "error", "batch_id":
self.batch_id, "message": str(e)})
return
self.log.warning("Polling batch %s failed (%s); retrying.",
self.batch_id, e)
diff --git a/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py
b/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py
index 487e32825aa..f5e0c3f422c 100644
--- a/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py
+++ b/providers/anthropic/tests/unit/anthropic/hooks/test_anthropic.py
@@ -38,6 +38,7 @@ from airflow.providers.anthropic.hooks.anthropic import (
SessionPollResult,
SessionStatus,
_create_session_error,
+ _is_permanent_poll_error,
build_budget,
evaluate_session_state,
validate_execute_complete_event,
@@ -45,7 +46,7 @@ from airflow.providers.anthropic.hooks.anthropic import (
pytest.importorskip("anthropic")
-from anthropic import AsyncAnthropic, BadRequestError,
WorkloadIdentityCredentials
+from anthropic import APIStatusError, AsyncAnthropic, BadRequestError,
WorkloadIdentityCredentials
from anthropic.types import BetaMonetaryAmount
from anthropic.types.beta import BetaManagedAgentsServerToolUsage,
BetaManagedAgentsSessionUsage
from anthropic.types.beta.beta_managed_agents_cache_creation_usage import (
@@ -1055,6 +1056,38 @@ class TestAnthropicHookFeatures:
hook.count_tokens([{"role": "user", "content": "x"}])
+def _api_status_error(status_code: int, headers: dict[str, str] | None = None)
-> APIStatusError:
+ # ``APIStatusError`` reads only these response attributes, so no HTTP
library is needed.
+ response = SimpleNamespace(status_code=status_code, headers=headers or {},
request=None)
+ return APIStatusError(f"HTTP {status_code}", response=response, body=None)
# type: ignore[arg-type]
+
+
+class TestIsPermanentPollError:
+ @pytest.mark.parametrize(
+ ("error", "permanent"),
+ [
+ pytest.param(_api_status_error(400), True, id="400"),
+ pytest.param(_api_status_error(401), True, id="401"),
+ pytest.param(_api_status_error(403), True, id="403"),
+ pytest.param(_api_status_error(404), True, id="404"),
+ pytest.param(_api_status_error(408), False,
id="408-request-timeout"),
+ pytest.param(_api_status_error(409), False, id="409-lock-timeout"),
+ pytest.param(_api_status_error(429), False, id="429-rate-limit"),
+ pytest.param(_api_status_error(500), False, id="500"),
+ pytest.param(_api_status_error(529), False, id="529-overloaded"),
+ pytest.param(
+ _api_status_error(400, {"x-should-retry": "true"}), False,
id="header-overrides-400"
+ ),
+ pytest.param(
+ _api_status_error(503, {"x-should-retry": "false"}), True,
id="header-overrides-503"
+ ),
+ pytest.param(RuntimeError("connection reset"), False,
id="no-http-status"),
+ ],
+ )
+ def test_matches_the_sdk_retry_rule(self, error, permanent):
+ assert _is_permanent_poll_error(error) is permanent
+
+
class TestWaitForBatch:
def _batch(self, status):
batch = mock.MagicMock()
@@ -1100,6 +1133,17 @@ class TestWaitForBatch:
AnthropicHook().wait_for_batch("batch_1", wait_seconds=0.01)
assert mock_get_batch.call_count == MAX_CONSECUTIVE_POLL_FAILURES
+ @mock.patch(f"{HOOK_PATH}.time.sleep", autospec=True)
+ @mock.patch.object(AnthropicHook, "get_batch", autospec=True)
+ @mock.patch.object(AnthropicHook, "get_connection", autospec=True)
+ def test_raises_permanent_api_error_without_retry(self,
mock_get_connection, mock_get_batch, mock_sleep):
+ mock_get_connection.return_value = _conn()
+ mock_get_batch.side_effect = _api_status_error(404)
+ with pytest.raises(APIStatusError, match="HTTP 404"):
+ AnthropicHook().wait_for_batch("batch_1", wait_seconds=0.01)
+ assert mock_get_batch.call_count == 1
+ mock_sleep.assert_not_called()
+
class TestTestConnection:
@mock.patch.object(AnthropicHook, "get_connection")
diff --git a/providers/anthropic/tests/unit/anthropic/triggers/test_batch.py
b/providers/anthropic/tests/unit/anthropic/triggers/test_batch.py
index 564681f545d..73820349d4b 100644
--- a/providers/anthropic/tests/unit/anthropic/triggers/test_batch.py
+++ b/providers/anthropic/tests/unit/anthropic/triggers/test_batch.py
@@ -17,9 +17,11 @@
from __future__ import annotations
import time
+from types import SimpleNamespace
from unittest import mock
import pytest
+from anthropic import APIStatusError
from airflow.providers.anthropic.triggers.batch import AnthropicBatchTrigger
from airflow.triggers.base import TriggerEvent
@@ -30,6 +32,12 @@ TRIGGER_PATH = "airflow.providers.anthropic.triggers.batch"
HOOK_PATH =
"airflow.providers.anthropic.hooks.anthropic.AnthropicHook.get_batch"
+def _api_status_error(status_code: int) -> APIStatusError:
+ # ``APIStatusError`` reads only these response attributes, so no HTTP
library is needed.
+ response = SimpleNamespace(status_code=status_code, headers={},
request=None)
+ return APIStatusError(f"HTTP {status_code}", response=response, body=None)
# type: ignore[arg-type]
+
+
def _batch(status, succeeded=0, errored=0, canceled=0, expired=0,
processing=0):
batch = mock.MagicMock()
batch.processing_status = status
@@ -109,6 +117,26 @@ class TestAnthropicBatchTrigger:
assert event == TriggerEvent({"status": "error", "batch_id":
self.BATCH_ID, "message": "kaboom"})
assert mock_get_batch.call_count == 5
+ @pytest.mark.asyncio
+ @mock.patch(f"{TRIGGER_PATH}.asyncio.sleep", autospec=True)
+ @mock.patch(HOOK_PATH, autospec=True)
+ async def test_permanent_api_error_yields_error_without_retry(self,
mock_get_batch, mock_sleep):
+ mock_get_batch.side_effect = _api_status_error(404)
+ event = await self._trigger().run().__anext__()
+ assert event == TriggerEvent({"status": "error", "batch_id":
self.BATCH_ID, "message": "HTTP 404"})
+ assert mock_get_batch.call_count == 1
+ mock_sleep.assert_not_awaited()
+
+ @pytest.mark.asyncio
+ @mock.patch(f"{TRIGGER_PATH}.asyncio.sleep", autospec=True)
+ @mock.patch(HOOK_PATH, autospec=True)
+ async def test_retryable_api_error_then_success(self, mock_get_batch,
mock_sleep):
+ mock_get_batch.side_effect = [_api_status_error(408), _batch("ended",
succeeded=1)]
+ event = await self._trigger().run().__anext__()
+ assert event.payload["status"] == "success"
+ assert mock_get_batch.call_count == 2
+ mock_sleep.assert_awaited_once_with(self.POLL)
+
@pytest.mark.asyncio
@mock.patch(f"{TRIGGER_PATH}.asyncio.sleep")
@mock.patch(HOOK_PATH)
