Adds sftp_sensor decorator

airflow/providers/sftp/decorators/__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/sftp/decorators/sensors/__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/sftp/decorators/sensors/sftp.py

@@ -0,0 +1,72 @@
+# 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 typing import Callable, Sequence
+
+from airflow.decorators.base import TaskDecorator, get_unique_task_id, task_decorator_factory
+from airflow.providers.sftp.sensors.sftp import SFTPSensor
+
+
+class _DecoratedSFTPSensorOperator(SFTPSensor):
+    """
+    Wraps a Python callable and captures args/kwargs when called for execution.
+
+    :param python_callable: A reference to an object that is callable
+    :param task_id: task Id
+    :param op_args: a list of positional arguments that will get unpacked when
+        calling your callable (templated)
+    :param op_kwargs: a dictionary of keyword arguments that will get unpacked
+        in your function (templated)
+    :param kwargs_to_upstream: For certain operators, we might need to upstream certain arguments
+        that would otherwise be absorbed by the DecoratedOperator (for example python_callable for the
+        PythonOperator). This gives a user the option to upstream kwargs as needed.
+    """
+
+    template_fields: Sequence[str] = ("op_args", "op_kwargs", *SFTPSensor.template_fields)
+
+    custom_operator_name = "@task.sftp_sensor"
+
+    # since we won't mutate the arguments, we should just do the shallow copy
+    # there are some cases we can't deepcopy the objects (e.g protobuf).
+    shallow_copy_attrs: Sequence[str] = ("python_callable",)
+
+    def __init__(
+        self,
+        *,
+        task_id: str,
+        **kwargs,
+    ) -> None:
+        kwargs.pop("multiple_outputs")
+        kwargs["task_id"] = get_unique_task_id(task_id, kwargs.get("dag"), kwargs.get("task_group"))
+        super().__init__(**kwargs)
+
+
+def sftp_sensor_task(python_callable: Callable | None = None, **kwargs) -> TaskDecorator:
+    """
+    Wraps a function into an Airflow operator.
+
+    Accepts kwargs for operator kwarg. Can be reused in a single DAG.
+    :param python_callable: Function to decorate
+    """
+    return task_decorator_factory(
+        python_callable=python_callable,
+        multiple_outputs=False,
+        decorated_operator_class=_DecoratedSFTPSensorOperator,
+        **kwargs,
+    )

airflow/providers/sftp/provider.yaml

@@ -68,6 +68,7 @@ sensors:
   - integration-name: SSH File Transfer Protocol (SFTP)
     python-modules:
       - airflow.providers.sftp.sensors.sftp
+      - airflow.providers.sftp.decorators.sensors.sftp
 
 hooks:
   - integration-name: SSH File Transfer Protocol (SFTP)
@@ -78,3 +79,7 @@ hooks:
 connection-types:
   - hook-class-name: airflow.providers.sftp.hooks.sftp.SFTPHook
     connection-type: sftp
+
+task-decorators:
+  - class-name: airflow.providers.sftp.decorators.sensors.sftp.sftp_sensor_task
+    name: sftp_sensor

airflow/providers/sftp/sensors/sftp.py

@@ -20,12 +20,12 @@
 
 import os
 from datetime import datetime
-from typing import TYPE_CHECKING, Sequence
+from typing import TYPE_CHECKING, Any, Callable, Sequence
 
 from paramiko.sftp import SFTP_NO_SUCH_FILE
 
 from airflow.providers.sftp.hooks.sftp import SFTPHook
-from airflow.sensors.base import BaseSensorOperator
+from airflow.sensors.base import BaseSensorOperator, PokeReturnValue
 from airflow.utils.timezone import convert_to_utc
 
 if TYPE_CHECKING:
@@ -54,6 +54,9 @@ def __init__(
         file_pattern: str = "",
         newer_than: datetime | None = None,
         sftp_conn_id: str = "sftp_default",
+        python_callable: Callable | None = None,
+        op_args: list | None = None,
+        op_kwargs: dict[str, Any] | None = None,
         **kwargs,
     ) -> None:
         super().__init__(**kwargs)
@@ -62,10 +65,14 @@ def __init__(
         self.hook: SFTPHook | None = None
         self.sftp_conn_id = sftp_conn_id
         self.newer_than: datetime | None = newer_than
+        self.python_callable: Callable | None = python_callable
+        self.op_args = op_args or []
+        self.op_kwargs = op_kwargs or {}
 
-    def poke(self, context: Context) -> bool:
+    def poke(self, context: Context) -> PokeReturnValue | bool:
         self.hook = SFTPHook(self.sftp_conn_id)
         self.log.info("Poking for %s, with pattern %s", self.path, self.file_pattern)
+        files_found = []
 
         if self.file_pattern:
             files_from_pattern = self.hook.get_files_by_pattern(self.path, self.file_pattern)
@@ -89,8 +96,18 @@ def poke(self, context: Context) -> bool:
                 _mod_time = convert_to_utc(datetime.strptime(mod_time, "%Y%m%d%H%M%S"))
                 _newer_than = convert_to_utc(self.newer_than)
                 if _newer_than <= _mod_time:
-                    return True
+                    files_found.append(actual_file_to_check)
             else:
-                return True
+                files_found.append(actual_file_to_check)
         self.hook.close_conn()
-        return False
+        if not len(files_found):
+            return False
+        if self.python_callable is not None:
+            if self.op_kwargs:
+                self.op_kwargs["files_found"] = files_found
+            callable_return = self.python_callable(*self.op_args, **self.op_kwargs)
+            return PokeReturnValue(
+                is_done=True,
+                xcom_value={"files_found": files_found, "decorator_return_value": callable_return},
+            )
+        return True

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

@@ -35,6 +35,7 @@
     :caption: References
 
     Connection types <connections/sftp>
+    Sensors <sensors/sftp_sensor>
     Python API <_api/airflow/providers/sftp/index>
 
 .. toctree::
@@ -45,6 +46,12 @@
     PyPI Repository <https://pypi.org/project/apache-airflow-providers-sftp/>
     Installing from sources <installing-providers-from-sources>
 
+.. toctree::
+    :hidden:
+    :maxdepth: 1
+    :caption: System tests
+
+    System Tests <_api/tests/system/providers/sftp/index>
 .. THE REMAINDER OF THE FILE IS AUTOMATICALLY GENERATED. IT WILL BE OVERWRITTEN AT RELEASE TIME!
 
 

docs/apache-airflow-providers-sftp/sensors/sftp_sensor.rst

@@ -0,0 +1,46 @@
+ .. 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.
+
+SFTP Sensor
+===========
+
+Looks for either a specific file or files with a specific pattern in a server using SFTP protocol.
+To get more information about this sensor visit :class:`~airflow.providers.sftp.sensors.sftp.SFTPSensor`
+
+.. exampleinclude:: /../../tests/system/providers/sftp/example_sftp_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_sftp_sensor]
+    :end-before: [END howto_operator_sftp_sensor]
+
+
+We can also use Taskflow API. It takes the same arguments as the :class:`~airflow.providers.sftp.sensors.sftp.SFTPSensor` along with -
+
+op_args (optional)
+    A list of positional arguments that will get unpacked when
+    calling your callable (templated)
+op_kwargs (optional)
+    A dictionary of keyword arguments that will get unpacked
+    in your function (templated)
+
+Whatever returned by the python callable is put into XCom.
+
+.. exampleinclude:: /../../tests/system/providers/sftp/example_sftp_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_operator_sftp_sensor_decorator]
+    :end-before: [END howto_operator_sftp_sensor_decorator]

tests/providers/sftp/decorators/__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.

tests/providers/sftp/decorators/sensors/__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.