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 57955085ecf Configure S3Hook transfer_config_args via service_config
(#65963)
57955085ecf is described below
commit 57955085ecfbd92567ac899406dc4269f966635d
Author: stephen-bracken <[email protected]>
AuthorDate: Tue Apr 28 22:12:33 2026 +0100
Configure S3Hook transfer_config_args via service_config (#65963)
---
providers/amazon/docs/connections/aws.rst | 23 ++++++++++++++++++++++
.../src/airflow/providers/amazon/aws/hooks/s3.py | 12 +++++++++--
.../amazon/tests/unit/amazon/aws/hooks/test_s3.py | 23 ++++++++++++++++++++++
3 files changed, 56 insertions(+), 2 deletions(-)
diff --git a/providers/amazon/docs/connections/aws.rst
b/providers/amazon/docs/connections/aws.rst
index ed5750d83ad..55ef6367326 100644
--- a/providers/amazon/docs/connections/aws.rst
+++ b/providers/amazon/docs/connections/aws.rst
@@ -383,6 +383,29 @@ provide selected options in the connection's extra field.
}
+.. _howto/connection:aws:s3-transfer-config:
+
+S3 Transfer Config
+------------------
+
+You can use the service config to set transfer config options in
:class:`~airflow.providers.amazon.aws.hooks.s3.S3Hook` methods.
+
+.. code-block:: json
+
+ {
+ "service_config": {
+ "s3": {
+ "transfer_config_args": {
+ "max_concurrency": 10,
+ "num_download_attempts": 10
+ }
+ }
+ }
+ }
+
+.. seealso::
+ - Boto3 TransferConfig:
:external:py:class:`boto3.s3.transfer.TransferConfig`
+
.. _howto/connection:aws:avoid-throttling-exceptions:
Avoid Throttling exceptions
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/hooks/s3.py
b/providers/amazon/src/airflow/providers/amazon/aws/hooks/s3.py
index f999bd4f165..5aba0e2b979 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/hooks/s3.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/hooks/s3.py
@@ -168,7 +168,8 @@ class S3Hook(AwsBaseHook):
Provide thick wrapper around :external+boto3:py:class:`boto3.client("s3")
<S3.Client>`
and :external+boto3:py:class:`boto3.resource("s3") <S3.ServiceResource>`.
- :param transfer_config_args: Configuration object for managed S3 transfers.
+ :param transfer_config_args: Default Configuration object for managed S3
transfers.
+ Can also be set in the service_config.
:param extra_args: Extra arguments that may be passed to the
download/upload operations.
.. seealso::
@@ -198,7 +199,7 @@ class S3Hook(AwsBaseHook):
if transfer_config_args and not isinstance(transfer_config_args, dict):
raise TypeError(f"transfer_config_args expected dict, got
{type(transfer_config_args).__name__}.")
- self.transfer_config = TransferConfig(**transfer_config_args or {})
+ self._transfer_config_args = transfer_config_args or {}
if extra_args and not isinstance(extra_args, dict):
raise TypeError(f"extra_args expected dict, got
{type(extra_args).__name__}.")
@@ -206,6 +207,13 @@ class S3Hook(AwsBaseHook):
super().__init__(*args, **kwargs)
+ @property
+ def transfer_config(self):
+ if not hasattr(self, "_transfer_config"):
+
self._transfer_config_args.update(self.service_config.get("transfer_config_args",
{}))
+ self._transfer_config =
TransferConfig(**self._transfer_config_args)
+ return self._transfer_config
+
@cached_property
def resource(self):
return self.get_session().resource(
diff --git a/providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py
b/providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py
index eea053fc52b..551e7739887 100644
--- a/providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py
+++ b/providers/amazon/tests/unit/amazon/aws/hooks/test_s3.py
@@ -100,6 +100,29 @@ class TestAwsS3Hook:
with pytest.raises(TypeError, match="transfer_config_args expected
dict, got .*"):
S3Hook(transfer_config_args=transfer_config_args)
+ @pytest.mark.parametrize(
+ ("transfer_config_args", "service_config"),
+ [
+ pytest.param(
+ {"use_threads": False},
+ {"transfer_config_args": {"use_threads": True}},
+ id="override_transfer_config_args",
+ ),
+ pytest.param(None, {"use_threads": True},
id="service_config_args"),
+ ],
+ )
+ def test_transfer_config_args_from_service_config(self,
transfer_config_args, service_config):
+ conn = Connection(
+ conn_id="s3_conn",
+ conn_type="aws",
+ extra={
+ "service_config": {"s3": service_config},
+ },
+ )
+ with mock.patch.dict("os.environ",
values={f"AIRFLOW_CONN_{conn.conn_id.upper()}": conn.get_uri()}):
+ hook = S3Hook(aws_conn_id=conn.conn_id,
transfer_config_args=transfer_config_args)
+ assert hook.transfer_config.use_threads is True
+
@pytest.mark.parametrize(
("url", "expected"),
[
