This is an automated email from the ASF dual-hosted git repository.
o-nikolas 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 6ae6994206a Stop deferred AWS tasks failing on a botocore Config
object (#74843)
6ae6994206a is described below
commit 6ae6994206a693a19ed06fd8a7b347045a9c84df
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Nov 6 22:56:16 2026 +0530
Stop deferred AWS tasks failing on a botocore Config object (#74843)
Amazon operators and sensors declare botocore_config as a dictionary, but
the hooks behind them also accept a botocore Config instance, and that form
works for ordinary execution: the client is built with the caller's settings
either way.
The deferred path is where the two diverge. The value is handed to the
trigger as an argument, trigger arguments are serialized before they are
stored, and serde cannot encode a Config. The task therefore fails at defer
time, which is after the operator has already made its AWS call, so the started
job keeps running with nothing left to watch it or report its result.
Resolving the parameter to the declared mapping in the one place where
these settings are assembled keeps the deferred and non deferred halves of a
task equivalent, so a configuration that works for one works for the other.
---
.../airflow/providers/amazon/aws/utils/mixins.py | 26 ++++++++-
.../tests/unit/amazon/aws/utils/test_mixins.py | 62 +++++++++++++++++++++-
2 files changed, 85 insertions(+), 3 deletions(-)
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/utils/mixins.py
b/providers/amazon/src/airflow/providers/amazon/aws/utils/mixins.py
index d2c86e3e4e3..b726a371bed 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/utils/mixins.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/utils/mixins.py
@@ -30,11 +30,33 @@ from __future__ import annotations
from functools import cache, cached_property
from typing import Any, Generic, NamedTuple, TypeVar, final
+from botocore.config import Config
+
from airflow.providers.amazon.aws.hooks.base_aws import AwsGenericHook
AwsHookType = TypeVar("AwsHookType", bound=AwsGenericHook)
+def normalize_botocore_config(
+ botocore_config: dict[str, Any] | Config | None,
+) -> dict[str, Any] | Config | None:
+ """
+ Reduce ``botocore_config`` to the mapping that the operator and sensor
interfaces declare.
+
+ A ``Config`` instance is accepted by the hooks but cannot be serialized,
so one that reaches a
+ trigger argument fails the task at defer time, once the operator's AWS
call has already been
+ made. Both forms build the same client, so the mapping is kept as the
single representation.
+ """
+ if not isinstance(botocore_config, Config):
+ return botocore_config
+ # Only the caller's own options, so that the mapping does not pin
whichever defaults the
+ # installed botocore happens to carry.
+ user_provided_options = getattr(botocore_config, "_user_provided_options",
None)
+ if user_provided_options is None:
+ return botocore_config
+ return dict(user_provided_options)
+
+
class AwsHookParams(NamedTuple):
"""
Default Aws Hook Parameters storage class.
@@ -53,7 +75,7 @@ class AwsHookParams(NamedTuple):
aws_conn_id: str | None,
region_name: str | None,
verify: bool | str | None,
- botocore_config: dict[str, Any] | None,
+ botocore_config: dict[str, Any] | Config | None,
additional_params: dict,
):
"""
@@ -83,7 +105,7 @@ class AwsHookParams(NamedTuple):
self.botocore_config = params.botocore_config
self.foo = foo
"""
- return cls(aws_conn_id, region_name, verify, botocore_config)
+ return cls(aws_conn_id, region_name, verify,
normalize_botocore_config(botocore_config))
class AwsBaseHookMixin(Generic[AwsHookType]):
diff --git a/providers/amazon/tests/unit/amazon/aws/utils/test_mixins.py
b/providers/amazon/tests/unit/amazon/aws/utils/test_mixins.py
index b127a9c83b0..e7e4d406016 100644
--- a/providers/amazon/tests/unit/amazon/aws/utils/test_mixins.py
+++ b/providers/amazon/tests/unit/amazon/aws/utils/test_mixins.py
@@ -17,8 +17,17 @@
from __future__ import annotations
import pytest
+from botocore.config import Config
-from airflow.providers.amazon.aws.utils.mixins import aws_template_fields
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.amazon.aws.operators.base_aws import AwsBaseOperator
+from airflow.providers.amazon.aws.utils.mixins import (
+ AwsHookParams,
+ aws_template_fields,
+ normalize_botocore_config,
+)
+
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_2_PLUS
def test_aws_template_fields():
@@ -38,3 +47,54 @@ def test_aws_template_fields():
def test_aws_template_fields_incorrect_types():
with pytest.raises(TypeError, match="Expected that all provided arguments
are strings"):
aws_template_fields(1, None, "test")
+
+
+class FakeBotocoreConfigOperator(AwsBaseOperator[S3Hook]):
+ aws_hook_class = S3Hook
+
+
+class TestNormalizeBotocoreConfig:
+ @pytest.mark.parametrize("value", [None, {}, {"read_timeout": 120,
"retries": {"max_attempts": 5}}])
+ def test_keeps_mapping_and_none_as_given(self, value):
+ assert normalize_botocore_config(value) is value
+
+ def test_reduces_config_object_to_the_options_it_was_given(self):
+ config = Config(read_timeout=120, retries={"max_attempts": 5},
region_name="eu-west-1")
+
+ assert normalize_botocore_config(config) == {
+ "read_timeout": 120,
+ "retries": {"max_attempts": 5},
+ "region_name": "eu-west-1",
+ }
+
+ def test_reduces_config_object_resolved_by_from_constructor(self):
+ params = AwsHookParams.from_constructor(
+ "aws_default", None, None, Config(read_timeout=120),
additional_params={}
+ )
+
+ assert params.botocore_config == {"read_timeout": 120}
+
+ def
test_keeps_config_object_when_botocore_withholds_the_given_options(self,
monkeypatch):
+ config = Config(read_timeout=120)
+ monkeypatch.delattr(type(config), "_user_provided_options",
raising=False)
+ monkeypatch.delattr(config, "_user_provided_options", raising=False)
+
+ assert normalize_botocore_config(config) is config
+
+ def test_operator_resolves_config_object_to_a_mapping(self):
+ op = FakeBotocoreConfigOperator(task_id="fake",
botocore_config=Config(read_timeout=120))
+
+ assert op.botocore_config == {"read_timeout": 120}
+
+ @pytest.mark.skipif(
+ not AIRFLOW_V_3_2_PLUS, reason="Trigger arguments go through
airflow.sdk.serde from Airflow 3.2.0"
+ )
+ def test_operator_settings_can_cross_the_trigger_boundary(self):
+ """A ``Config`` cannot cross serde, so the deferred half of the task
never receives one."""
+ from airflow.sdk.serde import serialize
+
+ op = FakeBotocoreConfigOperator(task_id="fake",
botocore_config=Config(read_timeout=120))
+
+ assert serialize({"botocore_config": op.botocore_config}) == {
+ "botocore_config": {"read_timeout": 120}
+ }
