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 dc939331f4c Add `S3ReadObjectOperator` (#65789)
dc939331f4c is described below
commit dc939331f4cd90892fd201931c466dffff977a4f
Author: John Jackson <[email protected]>
AuthorDate: Fri Apr 24 11:57:15 2026 -0700
Add `S3ReadObjectOperator` (#65789)
Add a new operator that reads an S3 object and returns its content
as a UTF-8 string. This is the read counterpart to the existing
`S3CreateObjectOperator`.
The operator:
- Extends `AwsBaseOperator[S3Hook]` following the standard pattern
- Supports both explicit `s3_bucket`/`s3_key` and full `s3://` URLs
- Delegates to `S3Hook.read_key()` for the actual read
- Returns the content as a string (auto-pushed to XCom)
- Includes OpenLineage support (input dataset facet)
Includes:
- Unit tests (moto integration + mock-based)
- System test DAG with infrastructure create/teardown
- Documentation with example snippets
---
providers/amazon/docs/operators/s3/s3.rst | 14 +++++
.../airflow/providers/amazon/aws/operators/s3.py | 64 ++++++++++++++++++++++
.../amazon/tests/system/amazon/aws/example_s3.py | 10 ++++
.../tests/unit/amazon/aws/operators/test_s3.py | 56 +++++++++++++++++++
4 files changed, 144 insertions(+)
diff --git a/providers/amazon/docs/operators/s3/s3.rst
b/providers/amazon/docs/operators/s3/s3.rst
index 2602eda66d8..d5bf4879606 100644
--- a/providers/amazon/docs/operators/s3/s3.rst
+++ b/providers/amazon/docs/operators/s3/s3.rst
@@ -178,6 +178,20 @@ for more information about Amazon S3 prefixes.
:start-after: [START howto_operator_s3_list_prefixes]
:end-before: [END howto_operator_s3_list_prefixes]
+.. _howto/operator:S3ReadObjectOperator:
+
+Read an Amazon S3 object
+========================
+
+To read the content of an Amazon S3 object and return it as a string you can
use
+:class:`~airflow.providers.amazon.aws.operators.s3.S3ReadObjectOperator`.
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_s3.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_s3_read_object]
+ :end-before: [END howto_operator_s3_read_object]
+
.. _howto/operator:S3ListOperator:
List Amazon S3 objects
diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py
b/providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py
index 50c1ac07983..2a5b12052b6 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py
@@ -930,3 +930,67 @@ class S3ListPrefixesOperator(AwsBaseOperator[S3Hook]):
)
return self.hook.list_prefixes(bucket_name=self.bucket,
prefix=self.prefix, delimiter=self.delimiter)
+
+
+class S3ReadObjectOperator(AwsBaseOperator[S3Hook]):
+ """
+ Read an S3 object and return its content as a string.
+
+ Uses ``S3Hook.read_key`` to fetch the object body, decode it as UTF-8,
+ and return the resulting string. The return value is automatically pushed
+ to XCom so downstream tasks can consume it.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:S3ReadObjectOperator`
+
+ :param s3_bucket: Name of the S3 bucket. (templated)
+ It should be omitted when ``s3_key`` is provided as a full ``s3://``
URL.
+ :param s3_key: The key of the object to read. (templated)
+ It can be either a full ``s3://`` style URL or a relative path from
root level.
+ When specified as a full ``s3://`` URL, omit ``s3_bucket``.
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
+ If this is ``None`` or empty then the default boto3 behaviour is used.
If
+ running Airflow in a distributed manner and aws_conn_id is None or
+ empty, then default boto3 configuration would be used (and must be
+ maintained on each worker node).
+ :param region_name: AWS region_name. If not specified then the default
boto3 behaviour is used.
+ :param verify: Whether or not to verify SSL certificates. See:
+
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
+ :param botocore_config: Configuration dictionary (key-values) for botocore
client. See:
+
https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html
+ """
+
+ template_fields: Sequence[str] = aws_template_fields("s3_bucket", "s3_key")
+ aws_hook_class = S3Hook
+
+ def __init__(
+ self,
+ *,
+ s3_bucket: str | None = None,
+ s3_key: str,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.s3_bucket = s3_bucket
+ self.s3_key = s3_key
+
+ def execute(self, context: Context) -> str:
+ bucket, key = self.hook.get_s3_bucket_key(self.s3_bucket, self.s3_key,
"s3_bucket", "s3_key")
+ self.log.info("Reading s3://%s/%s", bucket, key)
+ return self.hook.read_key(key=key, bucket_name=bucket)
+
+ def get_openlineage_facets_on_start(self):
+ from airflow.providers.common.compat.openlineage.facet import Dataset
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ bucket, key = self.hook.get_s3_bucket_key(self.s3_bucket, self.s3_key,
"s3_bucket", "s3_key")
+
+ input_dataset = Dataset(
+ namespace=f"s3://{bucket}",
+ name=key,
+ )
+
+ return OperatorLineage(
+ inputs=[input_dataset],
+ )
diff --git a/providers/amazon/tests/system/amazon/aws/example_s3.py
b/providers/amazon/tests/system/amazon/aws/example_s3.py
index 6ddf6d95bbd..184211e630d 100644
--- a/providers/amazon/tests/system/amazon/aws/example_s3.py
+++ b/providers/amazon/tests/system/amazon/aws/example_s3.py
@@ -30,6 +30,7 @@ from airflow.providers.amazon.aws.operators.s3 import (
S3ListOperator,
S3ListPrefixesOperator,
S3PutBucketTaggingOperator,
+ S3ReadObjectOperator,
)
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor,
S3KeysUnchangedSensor
from airflow.providers.common.compat.sdk import DAG, chain
@@ -141,6 +142,14 @@ with DAG(
replace=True,
)
+ # [START howto_operator_s3_read_object]
+ read_object = S3ReadObjectOperator(
+ task_id="read_object",
+ s3_bucket=bucket_name,
+ s3_key=key,
+ )
+ # [END howto_operator_s3_read_object]
+
# [START howto_operator_s3_list_prefixes]
list_prefixes = S3ListPrefixesOperator(
task_id="list_prefixes",
@@ -301,6 +310,7 @@ with DAG(
delete_tagging,
create_object,
create_object_2,
+ read_object,
list_prefixes,
list_keys,
[sensor_one_key, sensor_two_keys, sensor_key_with_function,
sensor_key_with_regex],
diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_s3.py
b/providers/amazon/tests/unit/amazon/aws/operators/test_s3.py
index a7501ee4bbf..f3a74a4dd3f 100644
--- a/providers/amazon/tests/unit/amazon/aws/operators/test_s3.py
+++ b/providers/amazon/tests/unit/amazon/aws/operators/test_s3.py
@@ -46,6 +46,7 @@ from airflow.providers.amazon.aws.operators.s3 import (
S3ListOperator,
S3ListPrefixesOperator,
S3PutBucketTaggingOperator,
+ S3ReadObjectOperator,
)
from airflow.providers.common.compat.openlineage.facet import (
Dataset,
@@ -1083,3 +1084,58 @@ class TestS3CreateObjectOperator:
def test_template_fields(self):
operator = S3CreateObjectOperator(task_id="test", s3_bucket="bucket",
s3_key="key", data="test")
validate_template_fields(operator)
+
+
+class TestS3ReadObjectOperator:
+ def setup_method(self):
+ self.operator = S3ReadObjectOperator(
+ task_id="test-s3-read-object",
+ s3_bucket=BUCKET_NAME,
+ s3_key=S3_KEY,
+ )
+
+ @mock_aws
+ def test_execute_reads_object(self):
+ conn = boto3.client("s3", region_name="us-east-1")
+ conn.create_bucket(Bucket=BUCKET_NAME)
+ conn.put_object(Bucket=BUCKET_NAME, Key=S3_KEY, Body=b"hello world
from s3")
+
+ result = self.operator.execute({})
+ assert result == "hello world from s3"
+
+ @mock_aws
+ def test_execute_with_s3_url(self):
+ conn = boto3.client("s3", region_name="us-east-1")
+ conn.create_bucket(Bucket=BUCKET_NAME)
+ conn.put_object(Bucket=BUCKET_NAME, Key=S3_KEY, Body=b"url-content")
+
+ op = S3ReadObjectOperator(
+ task_id="test-s3-url",
+ s3_key=f"s3://{BUCKET_NAME}/{S3_KEY}",
+ )
+ result = op.execute({})
+ assert result == "url-content"
+
+ @mock_aws
+ def test_execute_empty_object(self):
+ conn = boto3.client("s3", region_name="us-east-1")
+ conn.create_bucket(Bucket=BUCKET_NAME)
+ conn.put_object(Bucket=BUCKET_NAME, Key=S3_KEY, Body=b"")
+
+ result = self.operator.execute({})
+ assert result == ""
+
+ @pytest.mark.parametrize(("bucket", "key"), (("bucket", "file.txt"),
(None, "s3://bucket/file.txt")))
+ def test_get_openlineage_facets_on_start(self, bucket, key):
+ expected_input = Dataset(
+ namespace="s3://bucket",
+ name="file.txt",
+ )
+ op = S3ReadObjectOperator(task_id="test", s3_bucket=bucket, s3_key=key)
+ lineage = op.get_openlineage_facets_on_start()
+ assert len(lineage.outputs) == 0
+ assert len(lineage.inputs) == 1
+ assert lineage.inputs[0] == expected_input
+
+ def test_template_fields(self):
+ validate_template_fields(self.operator)
