feat: retrieve sftp file attrs onces instead multiple time

providers/src/airflow/providers/sftp/hooks/sftp.py

@@ -168,6 +168,15 @@ def list_directory(self, path: str) -> list[str]:
         files = sorted(conn.listdir(path))
         return files
 
+    def list_directory_with_attr(self, path: str) -> list[paramiko.SFTPAttributes]:
+        """
+        List files in a directory on the remote system including their SFTPAttributes.
+
+        :param path: full path to the remote directory to list
+        """
+        conn = self.get_conn()
+        return [file for file in conn.listdir_attr(path)]
+
     def mkdir(self, path: str, mode: int = 0o777) -> None:
         """
         Create a directory on the remote system.
@@ -344,10 +353,9 @@ def walktree(
             (form: ``func(str)``)
         :param bool recurse: *Default: True* - should it recurse
         """
-        conn = self.get_conn()
-        for entry in self.list_directory(path):
-            pathname = os.path.join(path, entry)
-            mode = conn.stat(pathname).st_mode
+        for entry in self.list_directory_with_attr(path):
+            pathname = os.path.join(path, entry.filename)
+            mode = entry.st_mode
             if stat.S_ISDIR(mode):  # type: ignore
                 # It's a directory, call the dcallback function
                 dcallback(pathname)
@@ -423,9 +431,9 @@ def get_files_by_pattern(self, path, fnmatch_pattern) -> list[str]:
         :return: list of string containing the found files, or an empty list if none matched
         """
         matched_files = []
-        for file in self.list_directory(path):
-            if fnmatch(file, fnmatch_pattern):
-                matched_files.append(file)
+        for file in self.list_directory_with_attr(path):
+            if fnmatch(file.filename, fnmatch_pattern):
+                matched_files.append(file.filename)
 
         return matched_files
 

providers/tests/sftp/hooks/test_sftp.py

@@ -117,6 +117,12 @@ def test_list_directory(self):
         output = self.hook.list_directory(path=os.path.join(self.temp_dir, TMP_DIR_FOR_TESTS))
         assert output == [SUB_DIR, FIFO_FOR_TESTS]
 
+    def test_list_directory_with_attr(self):
+        output = self.hook.list_directory_with_attr(path=os.path.join(self.temp_dir, TMP_DIR_FOR_TESTS))
+        file_names = [f.filename for f in output]
+        assert all(isinstance(f, paramiko.SFTPAttributes) for f in output)
+        assert sorted(file_names) == [SUB_DIR, FIFO_FOR_TESTS]
+
     def test_mkdir(self):
         new_dir_name = "mk_dir"
         self.hook.mkdir(os.path.join(self.temp_dir, TMP_DIR_FOR_TESTS, new_dir_name))
@@ -457,7 +463,7 @@ def test_get_matched_files_several_pattern(self):
 
     def test_get_all_matched_files(self):
         output = self.hook.get_files_by_pattern(self.temp_dir, "test_*.txt")
-        assert output == [TMP_FILE_FOR_TESTS, ANOTHER_FILE_FOR_TESTS]
+        assert sorted(output) == [TMP_FILE_FOR_TESTS, ANOTHER_FILE_FOR_TESTS]
 
     def test_get_matched_files_with_different_pattern(self):
         output = self.hook.get_files_by_pattern(self.temp_dir, "*_file_*.txt")