Add documentation and sample DAG for ExasolToS3Operator

airflow-core/tests/unit/always/test_project_structure.py

@@ -514,8 +514,6 @@ class TestAmazonProviderProjectStructure(ExampleCoverageTest):
     }
 
     MISSING_EXAMPLES_FOR_CLASSES = {
-        # S3 Exasol transfer difficult to test, see: https://github.com/apache/airflow/issues/22632
-        "airflow.providers.amazon.aws.transfers.exasol_to_s3.ExasolToS3Operator",
         # These operations take a lot of time, there are commented out in the system tests for this reason
         "airflow.providers.amazon.aws.operators.dms.DmsStartReplicationOperator",
         "airflow.providers.amazon.aws.operators.dms.DmsStopReplicationOperator",

providers/amazon/docs/transfer/exasol_to_s3.rst

@@ -0,0 +1,63 @@
+ .. 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.
+
+===================
+Exasol to Amazon S3
+===================
+
+Use the ``ExasolToS3Operator`` transfer to export data from an `Exasol <https://www.exasol.com/>`__ database
+table or query result into an Amazon Simple Storage Service (S3) file.
+
+Prerequisite Tasks
+------------------
+
+.. include:: ../_partials/prerequisite_tasks.rst
+
+Operators
+---------
+
+.. _howto/operator:ExasolToS3Operator:
+
+Exasol To Amazon S3 transfer operator
+=====================================
+
+This operator exports data from an Exasol database to an Amazon S3 file.
+It uses the ``export_to_file`` method of the underlying
+`pyexasol <https://exasol.github.io/pyexasol/>`__ connection to export the data to a local
+temporary file, and then uploads the file to S3 with the
+:class:`~airflow.providers.amazon.aws.hooks.s3.S3Hook`.
+
+Use the ``query_or_table`` parameter to export either the result of a SQL query or a whole table.
+Extra parameters can be passed to the underlying export method with ``query_params`` and
+``export_params``.
+
+To get more information about this operator visit:
+:class:`~airflow.providers.amazon.aws.transfers.exasol_to_s3.ExasolToS3Operator`
+
+Example usage:
+
+.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_exasol_to_s3.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_transfer_exasol_to_s3]
+    :end-before: [END howto_transfer_exasol_to_s3]
+
+Reference
+---------
+
+* `pyexasol library documentation <https://exasol.github.io/pyexasol/>`__
+* `AWS boto3 library documentation for Amazon S3 <https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html>`__

providers/amazon/provider.yaml

@@ -979,6 +979,7 @@ transfers:
     python-module: airflow.providers.amazon.aws.transfers.s3_to_ftp
   - source-integration-name: Exasol
     target-integration-name: Amazon Simple Storage Service (S3)
+    how-to-guide: /docs/apache-airflow-providers-amazon/transfer/exasol_to_s3.rst
     python-module: airflow.providers.amazon.aws.transfers.exasol_to_s3
   - source-integration-name: File Transfer Protocol (FTP)
     target-integration-name: Amazon Simple Storage Service (S3)

providers/amazon/src/airflow/providers/amazon/aws/transfers/exasol_to_s3.py

@@ -35,7 +35,10 @@ class ExasolToS3Operator(BaseOperator):
     """
     Export data from Exasol database to AWS S3 bucket.
 
-    :param query_or_table: the sql statement to be executed or table name to export
+    :param query_or_table: the sql statement to be executed or table name to export.
+        A table name must be a single plain identifier (e.g. ``EMPLOYEES``): dotted
+        names such as ``SCHEMA.TABLE`` are rejected by ``pyexasol`` as unsafe
+        identifiers, so use a query instead (e.g. ``SELECT * FROM SCHEMA.TABLE``).
     :param key: S3 key that will point to the file
     :param bucket_name: Name of the bucket in which to store the file
     :param replace: A flag to decide whether or not to overwrite the key

providers/amazon/src/airflow/providers/amazon/get_provider_info.py

@@ -1133,6 +1133,7 @@ def get_provider_info():
             {
                 "source-integration-name": "Exasol",
                 "target-integration-name": "Amazon Simple Storage Service (S3)",
+                "how-to-guide": "/docs/apache-airflow-providers-amazon/transfer/exasol_to_s3.rst",
                 "python-module": "airflow.providers.amazon.aws.transfers.exasol_to_s3",
             },
             {

providers/amazon/tests/system/amazon/aws/example_exasol_to_s3.py

@@ -0,0 +1,92 @@
+# 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 airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator, S3DeleteBucketOperator
+from airflow.providers.amazon.aws.transfers.exasol_to_s3 import ExasolToS3Operator
+from airflow.providers.common.compat.sdk import DAG, chain
+
+try:
+    from airflow.sdk import TriggerRule
+except ImportError:
+    # Compatibility for Airflow < 3.1
+    from airflow.utils.trigger_rule import TriggerRule  # type: ignore[no-redef,attr-defined]
+from airflow.providers.common.compat.sdk import timezone
+
+from system.amazon.aws.utils import SystemTestContextBuilder
+
+DAG_ID = "example_exasol_to_s3"
+
+# Externally fetched variables:
+# The table (or view) exported below. The connection used by exasol_conn_id
+# (default: exasol_default) must point to a running Exasol instance.
+EXASOL_TABLE_KEY = "EXASOL_TABLE"
+
+sys_test_context_task = SystemTestContextBuilder().add_variable(EXASOL_TABLE_KEY).build()
+
+with DAG(
+    DAG_ID,
+    schedule="@once",
+    start_date=timezone.datetime(2021, 1, 1),
+    catchup=False,
+) as dag:
+    test_context = sys_test_context_task()
+    env_id = test_context["ENV_ID"]
+    exasol_table = test_context[EXASOL_TABLE_KEY]
+
+    s3_bucket = f"{env_id}-exasol-to-s3-bucket"
+    s3_key = f"{env_id}-exasol-to-s3-key"
+
+    create_s3_bucket = S3CreateBucketOperator(task_id="create_s3_bucket", bucket_name=s3_bucket)
+
+    # [START howto_transfer_exasol_to_s3]
+    exasol_to_s3_job = ExasolToS3Operator(
+        task_id="exasol_to_s3_job",
+        query_or_table=exasol_table,
+        key=s3_key,
+        bucket_name=s3_bucket,
+        replace=True,
+    )
+    # [END howto_transfer_exasol_to_s3]
+
+    delete_s3_bucket = S3DeleteBucketOperator(
+        task_id="delete_s3_bucket",
+        bucket_name=s3_bucket,
+        force_delete=True,
+        trigger_rule=TriggerRule.ALL_DONE,
+    )
+
+    chain(
+        # TEST SETUP
+        test_context,
+        create_s3_bucket,
+        # TEST BODY
+        exasol_to_s3_job,
+        # TEST TEARDOWN
+        delete_s3_bucket,
+    )
+
+    from tests_common.test_utils.watcher import watcher
+
+    # This test needs watcher in order to properly mark success/failure
+    # when "tearDown" task with trigger rule is part of the DAG
+    list(dag.tasks) >> watcher()
+
+from tests_common.test_utils.system_tests import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: contributing-docs/testing/system_tests.rst)
+test_run = get_test_run(dag)