Add FTPFileTransmitOperator (#26974)

* Add FTPFileTransmitOperator

airflow/providers/ftp/hooks/ftp.py

@@ -67,7 +67,7 @@ def get_conn(self) -> ftplib.FTP:
     def close_conn(self):
         """
         Closes the connection. An error will occur if the
-        connection wasn't ever opened.
+        connection was not ever opened.
         """
         conn = self.conn
         conn.quit()

airflow/providers/ftp/operators/__init__.py

@@ -0,0 +1,16 @@
+# 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.

airflow/providers/ftp/operators/ftp.py

@@ -0,0 +1,132 @@
+#
+# 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.
+"""This module contains FTP operator."""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any, Sequence
+
+from airflow.compat.functools import cached_property
+from airflow.models import BaseOperator
+from airflow.providers.ftp.hooks.ftp import FTPHook
+
+
+class FTPOperation:
+    """Operation that can be used with FTP"""
+
+    PUT = "put"
+    GET = "get"
+
+
+class FTPFileTransmitOperator(BaseOperator):
+    """
+    FTPFileTransmitOperator for transferring files from remote host to local or vice a versa.
+    This operator uses an FTPHook to open ftp transport channel that serve as basis
+    for file transfer.
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:FTPFileTransmitOperator`
+
+    :param ftp_conn_id: :ref:`ftp connection id<howto/connection:ftp>`
+        from airflow Connections.
+    :param local_filepath: local file path to get or put. (templated)
+    :param remote_filepath: remote file path to get or put. (templated)
+    :param operation: specify operation 'get' or 'put', defaults to put
+    :param create_intermediate_dirs: create missing intermediate directories when
+        copying from remote to local and vice-versa. Default is False.
+
+        Example: The following task would copy ``file.txt`` to the remote host
+        at ``/tmp/tmp1/tmp2/`` while creating ``tmp``,``tmp1`` and ``tmp2`` if they
+        don't exist. If the ``create_intermediate_dirs`` parameter is not passed it would error
+        as the directory does not exist. ::
+
+            put_file = FTPFileTransmitOperator(
+                task_id="test_ftp",
+                ftp_conn_id="ftp_default",
+                local_filepath="/tmp/file.txt",
+                remote_filepath="/tmp/tmp1/tmp2/file.txt",
+                operation="put",
+                create_intermediate_dirs=True,
+                dag=dag
+            )
+    """
+
+    template_fields: Sequence[str] = ("local_filepath", "remote_filepath")
+
+    def __init__(
+        self,
+        *,
+        ftp_conn_id: str = "ftp_default",
+        local_filepath: str | list[str],
+        remote_filepath: str | list[str],
+        operation: str = FTPOperation.PUT,
+        create_intermediate_dirs: bool = False,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.ftp_conn_id = ftp_conn_id
+        self.operation = operation
+        self.create_intermediate_dirs = create_intermediate_dirs
+
+        if isinstance(local_filepath, str):
+            self.local_filepath = [local_filepath]
+        else:
+            self.local_filepath = local_filepath
+
+        if isinstance(remote_filepath, str):
+            self.remote_filepath = [remote_filepath]
+        else:
+            self.remote_filepath = remote_filepath
+
+        if len(self.local_filepath) != len(self.remote_filepath):
+            raise ValueError(
+                f"{len(self.local_filepath)} paths in local_filepath "
+                f"!= {len(self.remote_filepath)} paths in remote_filepath"
+            )
+
+        if self.operation.lower() not in [FTPOperation.GET, FTPOperation.PUT]:
+            raise TypeError(
+                f"Unsupported operation value {self.operation}, "
+                f"expected {FTPOperation.GET} or {FTPOperation.PUT}."
+            )
+
+    @cached_property
+    def hook(self) -> FTPHook:
+        """Create and return an FTPHook."""
+        return FTPHook(ftp_conn_id=self.ftp_conn_id)
+
+    def execute(self, context: Any) -> str | list[str] | None:
+        file_msg = None
+        for local_filepath, remote_filepath in zip(self.local_filepath, self.remote_filepath):
+            if self.operation.lower() == FTPOperation.GET:
+                local_folder = os.path.dirname(local_filepath)
+                if self.create_intermediate_dirs:
+                    Path(local_folder).mkdir(parents=True, exist_ok=True)
+                file_msg = f"from {remote_filepath} to {local_filepath}"
+                self.log.info("Starting to transfer %s", file_msg)
+                self.hook.retrieve_file(remote_filepath, local_filepath)
+            else:
+                remote_folder = os.path.dirname(remote_filepath)
+                if self.create_intermediate_dirs:
+                    self.hook.create_directory(remote_folder)
+                file_msg = f"from {local_filepath} to {remote_filepath}"
+                self.log.info("Starting to transfer file %s", file_msg)
+                self.hook.store_file(remote_filepath, local_filepath)
+        return self.local_filepath

airflow/providers/ftp/provider.yaml

@@ -42,6 +42,11 @@ integrations:
     logo: /integration-logos/ftp/FTP.png
     tags: [protocol]
 
+operators:
+  - integration-name: File Transfer Protocol (FTP)
+    python-modules:
+      - airflow.providers.ftp.operators.ftp
+
 sensors:
   - integration-name: File Transfer Protocol (FTP)
     python-modules:

airflow/utils/db.py

@@ -311,6 +311,18 @@ def create_default_connections(session: Session = NEW_SESSION):
         ),
         session,
     )
+    merge_conn(
+        Connection(
+            conn_id="ftp_default",
+            conn_type="ftp",
+            host="localhost",
+            port=21,
+            login="airflow",
+            password="airflow",
+            extra='{"key_file": "~/.ssh/id_rsa", "no_host_key_check": true}',
+        ),
+        session,
+    )
     merge_conn(
         Connection(
             conn_id="google_cloud_default",

docs/apache-airflow-providers-ftp/index.rst

@@ -29,10 +29,17 @@ Content
     Connection Types <connections/ftp>
     Python API <_api/airflow/providers/ftp/index>
 
+.. toctree::
+    :maxdepth: 1
+    :caption: Operators
+
+    Operators <operators/index>
+
 .. toctree::
     :maxdepth: 1
     :caption: Resources
 
+    Example DAGs <https://github.com/apache/airflow/tree/main/tests/system/providers/ftp/example_ftp.py>
     PyPI Repository <https://pypi.org/project/apache-airflow-providers-ftp/>
     Installing from sources <installing-providers-from-sources>
 
@@ -45,6 +52,11 @@ Content
 
     Detailed list of commits <commits>
 
+.. toctree::
+    :hidden:
+    :caption: System tests
+
+    System Tests <_api/tests/system/providers/ftp/index>
 
 Package apache-airflow-providers-ftp
 ------------------------------------------------------

docs/apache-airflow-providers-ftp/operators/index.rst

@@ -0,0 +1,48 @@
+.. 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.
+
+
+
+.. _howto/operator:FTPFileTransmitOperator:
+
+FTPFileTransmitOperator
+=========================
+
+
+Use the FTPFileTransmitOperator to get or
+pull files to/from an FTP server.
+
+Using the Operator
+^^^^^^^^^^^^^^^^^^
+
+For parameter definition take a look at :class:`~airflow.providers.ftp.operators.FTPFileTransmitOperator`.
+
+The below example shows how to use the FTPFileTransmitOperator to transfer a locally stored file to a remote FTP Server:
+
+.. exampleinclude:: /../../tests/system/providers/ftp/example_ftp.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_ftp_put]
+    :end-before: [END howto_operator_ftp_put]
+
+The below example shows how to use the FTPFileTransmitOperator to pull a file from a remote FTP Server.
+
+.. exampleinclude:: /../../tests/system/providers/ftp/example_ftp.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_ftp_get]
+    :end-before: [END howto_operator_ftp_get]

tests/providers/ftp/operators/__init__.py

@@ -0,0 +1,16 @@
+# 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.