Fix SFTPSensor.newer_than not working with jinja logical ds/ts expression

airflow/providers/sftp/sensors/sftp.py

@@ -21,6 +21,7 @@
 
 import os
 from datetime import datetime, timedelta
+from dateutil.parser import parse as parse_date
 from typing import TYPE_CHECKING, Any, Callable, Sequence
 
 from paramiko.sftp import SFTP_NO_SUCH_FILE
@@ -57,7 +58,7 @@ def __init__(
         *,
         path: str,
         file_pattern: str = "",
-        newer_than: datetime | None = None,
+        newer_than: datetime | str | None = None,
         sftp_conn_id: str = "sftp_default",
         python_callable: Callable | None = None,
         op_args: list | None = None,
@@ -70,7 +71,7 @@ def __init__(
         self.file_pattern = file_pattern
         self.hook: SFTPHook | None = None
         self.sftp_conn_id = sftp_conn_id
-        self.newer_than: datetime | None = newer_than
+        self.newer_than: datetime | str | None = newer_than
         self.python_callable: Callable | None = python_callable
         self.op_args = op_args or []
         self.op_kwargs = op_kwargs or {}
@@ -105,6 +106,8 @@ def poke(self, context: Context) -> PokeReturnValue | bool:
                 continue
 
             if self.newer_than:
+                if isinstance(self.newer_than, str):
+                    self.newer_than = parse_date(self.newer_than)
                 _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:

tests/providers/sftp/sensors/test_sftp.py

@@ -108,6 +108,17 @@ def test_naive_datetime(self, sftp_hook_mock):
         sftp_hook_mock.return_value.get_mod_time.assert_called_once_with("/path/to/file/1970-01-01.txt")
         assert not output
 
+    @patch("airflow.providers.sftp.sensors.sftp.SFTPHook")
+    def test_naive_datetime_in_pattern(self, sftp_hook_mock):
+        sftp_hook_mock.return_value.get_mod_time.return_value = "19700101000000"
+        sftp_sensor = SFTPSensor(
+            task_id="unit_test", path="/path/to/file/1970-01-01.txt", newer_than="2020-01-02 00:00:00+00:00"
+        )
+        context = {"ds": "1970-01-00"}
+        output = sftp_sensor.poke(context)
+        sftp_hook_mock.return_value.get_mod_time.assert_called_once_with("/path/to/file/1970-01-01.txt")
+        assert not output
+
     @patch("airflow.providers.sftp.sensors.sftp.SFTPHook")
     def test_file_present_with_pattern(self, sftp_hook_mock):
         sftp_hook_mock.return_value.get_mod_time.return_value = "19700101000000"