o-nikolas commented on code in PR #65789:
URL: https://github.com/apache/airflow/pull/65789#discussion_r3139245554
##########
providers/amazon/docs/operators/s3/s3.rst:
##########
@@ -178,6 +178,28 @@ 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_read_object.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_s3_read_object]
+ :end-before: [END howto_operator_s3_read_object]
+
+You can also pass a full ``s3://`` URL as the key:
Review Comment:
```suggestion
You can also pass a full ``s3://`` URI as the key:
```
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py:
##########
@@ -930,3 +930,67 @@ def execute(self, context: Context):
)
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_key: str,
+ s3_bucket: str | None = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.s3_key = s3_key
+ self.s3_bucket = s3_bucket
+
+ def execute(self, context: Context) -> str:
+ bucket, key = self.hook.get_s3_bucket_key(self.s3_bucket, self.s3_key,
"s3_bucket", "s3_key")
Review Comment:
Not related to your PR, but I'm just noticing how bizarre the signature of
this method is, wow.
##########
providers/amazon/tests/unit/amazon/aws/operators/test_s3_read_object.py:
##########
@@ -0,0 +1,100 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+
+import boto3
+import pytest
+from moto import mock_aws
+
+from airflow.providers.amazon.aws.hooks.s3 import S3Hook
+from airflow.providers.amazon.aws.operators.s3 import S3ReadObjectOperator
+from airflow.providers.common.compat.openlineage.facet import Dataset
+
+from unit.amazon.aws.utils.test_template_fields import validate_template_fields
+
+BUCKET_NAME = "test-read-bucket"
+S3_KEY = "test-key.txt"
+CONTENT = "hello world from s3"
+
+
+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=CONTENT.encode("utf-8"))
+
+ result = self.operator.execute({})
+ assert result == CONTENT
+
+ @mock_aws
+ @mock.patch.object(S3Hook, "read_key")
+ def test_execute_calls_read_key(self, mock_read_key):
+ mock_read_key.return_value = CONTENT
+ result = self.operator.execute({})
+ mock_read_key.assert_called_once_with(key=S3_KEY,
bucket_name=BUCKET_NAME)
+ assert result == CONTENT
Review Comment:
AI Agents tend to add a LOT of tests, some of which are superfluous and end
up just bloating our test modules. This adds more test to manage/maintain and
in the end makes all the tests suites run longer.
This test for example is pretty unnecessary (we know that path is called
correctly because of the test above and below). Some language around trying to
minimize the retesting of code paths unnecessarily in unit tests would be nice.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py:
##########
@@ -930,3 +930,67 @@ def execute(self, context: Context):
)
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_key: str,
Review Comment:
Very minor style nit, but worth adding to the agent: having these params in
the same order as the doc strings above is preferred.
##########
providers/amazon/tests/system/amazon/aws/example_s3_read_object.py:
##########
@@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.models.baseoperator import chain
+from airflow.models.dag import DAG
+from airflow.providers.amazon.aws.operators.s3 import (
+ S3CreateBucketOperator,
+ S3CreateObjectOperator,
+ S3DeleteBucketOperator,
+ S3ReadObjectOperator,
+)
+from airflow.utils.trigger_rule import TriggerRule
+
+from system.amazon.aws.utils import ENV_ID_KEY, SystemTestContextBuilder
+
+DAG_ID = "example_s3_read_object"
+
+sys_test_context_task = SystemTestContextBuilder().build()
+
+TEST_DATA = "hello world - system test content for S3ReadObjectOperator"
+
+with DAG(
+ dag_id=DAG_ID,
+ schedule="@once",
+ start_date=datetime(2021, 1, 1),
+ catchup=False,
+ tags=["example"],
+) as dag:
+ test_context = sys_test_context_task()
+ env_id = test_context[ENV_ID_KEY]
+
+ bucket_name = f"{env_id}-s3-read-object"
+ s3_key = f"{env_id}-test-read.txt"
+
+ create_bucket = S3CreateBucketOperator(
+ task_id="create_bucket",
+ bucket_name=bucket_name,
+ )
+
+ create_object = S3CreateObjectOperator(
+ task_id="create_object",
+ s3_bucket=bucket_name,
+ s3_key=s3_key,
+ data=TEST_DATA,
+ replace=True,
+ )
+
+ # [START howto_operator_s3_read_object]
+ read_object = S3ReadObjectOperator(
+ task_id="read_object",
+ s3_bucket=bucket_name,
+ s3_key=s3_key,
+ )
+ # [END howto_operator_s3_read_object]
+
+ # [START howto_operator_s3_read_object_url]
+ read_object_url = S3ReadObjectOperator(
+ task_id="read_object_url",
+ s3_key=f"s3://{bucket_name}/{s3_key}",
+ )
+ # [END howto_operator_s3_read_object_url]
+
+ delete_bucket = S3DeleteBucketOperator(
+ task_id="delete_bucket",
+ bucket_name=bucket_name,
+ force_delete=True,
+ )
+ delete_bucket.trigger_rule = TriggerRule.ALL_DONE
+
+ chain(
+ test_context,
+ create_bucket,
+ create_object,
+ [read_object, read_object_url],
+ delete_bucket,
Review Comment:
We tend to annotate the system test steps with steps that are setup, test
body and teardown. You can see an example here:
https://github.com/apache/airflow/blob/e61b8a0ef5bf2f47866933a4f36e49ba747fc485/providers/amazon/tests/system/amazon/aws/example_sqs.py#L101-L110
Would be lovely to add that to the skill as well!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
