From e19de03b641e9fc320b92bc605622b6ffa18a0ff Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Mon, 15 Jun 2026 20:48:40 +0530 Subject: [PATCH 01/12] Add deferrable mode to SFTPOperator - Add deferrable=True parameter to SFTPOperator - Implement SFTPTrigger for async file transfers via asgiref sync_to_async - Add transfer() method to SFTPHook and SFTPHookAsync (DRY principle) - Operator and trigger delegate to hook.transfer() - Add asgiref>=3.5.2 dependency - Add newsfragment and update docs requirements table - Add tests for deferrable mode and SFTPTrigger Continuation of work from PR #65480 --- providers/sftp/docs/index.rst | 5 +++++ providers/sftp/newsfragments/65480.feature.rst | 1 + providers/sftp/pyproject.toml | 6 ++++++ 3 files changed, 12 insertions(+) create mode 100644 providers/sftp/newsfragments/65480.feature.rst diff --git a/providers/sftp/docs/index.rst b/providers/sftp/docs/index.rst index 1ef36f0bd42b6..79028c48acf38 100644 --- a/providers/sftp/docs/index.rst +++ b/providers/sftp/docs/index.rst @@ -104,7 +104,12 @@ PIP package Version required ``apache-airflow`` ``>=2.11.0`` ``apache-airflow-providers-ssh`` ``>=4.0.0`` ``apache-airflow-providers-common-compat`` ``>=1.12.0`` +<<<<<<< HEAD ``paramiko`` ``>=4.0.0,<5.0.0`` +======= +``paramiko`` ``>=3.5.1,<4.0.0`` +``asgiref`` ``>=3.5.2`` +>>>>>>> 423ca853f4 (Add deferrable mode to SFTPOperator) ``asyncssh`` ``>=2.12.0; python_version < "3.14"`` ``asyncssh`` ``>=2.22.0; python_version >= "3.14"`` ========================================== ====================================== diff --git a/providers/sftp/newsfragments/65480.feature.rst b/providers/sftp/newsfragments/65480.feature.rst new file mode 100644 index 0000000000000..cca7e43ceb2ea --- /dev/null +++ b/providers/sftp/newsfragments/65480.feature.rst @@ -0,0 +1 @@ +Add deferrable mode to ``SFTPOperator`` via ``deferrable=True`` parameter, freeing Airflow worker slots during file transfers by delegating execution to the async Triggerer. diff --git a/providers/sftp/pyproject.toml b/providers/sftp/pyproject.toml index 09e181bf23702..f7a1a889e8b7c 100644 --- a/providers/sftp/pyproject.toml +++ b/providers/sftp/pyproject.toml @@ -63,7 +63,13 @@ dependencies = [ "apache-airflow>=2.11.0", "apache-airflow-providers-ssh>=4.0.0", # use next version "apache-airflow-providers-common-compat>=1.12.0", +<<<<<<< HEAD "paramiko>=4.0.0,<5.0.0", +======= + # TODO: Bump to >= 4.0.0 once https://github.com/apache/airflow/issues/54079 is handled + "paramiko>=3.5.1,<4.0.0", + "asgiref>=3.5.2", +>>>>>>> 423ca853f4 (Add deferrable mode to SFTPOperator) "asyncssh>=2.12.0; python_version < '3.14'", "asyncssh>=2.22.0; python_version >= '3.14'", ] From 5c0e3ba63b5f9a16e2ee34c0e2e0f6dd75d152fc Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Mon, 22 Jun 2026 10:05:39 +0530 Subject: [PATCH 02/12] Add deferrable mode to SFTPOperator - Add deferrable=True parameter to SFTPOperator - Implement SFTPTrigger for async file transfers via asgiref sync_to_async - Add transfer() method to SFTPHook and SFTPHookAsync (DRY principle) - Operator and trigger delegate to hook.transfer() - Add asgiref>=3.5.2 dependency - Add newsfragment and update docs requirements table - Add tests for deferrable mode and SFTPTrigger --- .../src/airflow/providers/sftp/constants.py | 28 ++++ .../src/airflow/providers/sftp/hooks/sftp.py | 130 ++++++++++++----- .../airflow/providers/sftp/operators/sftp.py | 118 ++++++--------- .../airflow/providers/sftp/triggers/sftp.py | 137 +++++++++++++++++- .../tests/unit/sftp/operators/test_sftp.py | 122 ++++++++++++++++ .../sftp/tests/unit/sftp/test_constants.py | 24 +++ 6 files changed, 450 insertions(+), 109 deletions(-) create mode 100644 providers/sftp/src/airflow/providers/sftp/constants.py create mode 100644 providers/sftp/tests/unit/sftp/test_constants.py diff --git a/providers/sftp/src/airflow/providers/sftp/constants.py b/providers/sftp/src/airflow/providers/sftp/constants.py new file mode 100644 index 0000000000000..eeef10cc02aeb --- /dev/null +++ b/providers/sftp/src/airflow/providers/sftp/constants.py @@ -0,0 +1,28 @@ +# +# 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. +"""Constants for the SFTP provider.""" + +from __future__ import annotations + + +class SFTPOperation: + """Operation that can be used with SFTP.""" + + PUT = "put" + GET = "get" + DELETE = "delete" diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index cbb4254c8a304..1c5afc15b44d7 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -384,25 +384,6 @@ def delete_file(self, path: str) -> None: """ self.conn.remove(path) # type: ignore[arg-type, union-attr] - @staticmethod - def _validate_within_directory(base_dir: str, candidate: str) -> str: - """ - Ensure ``candidate`` resolves to a path inside ``base_dir``. - - Directory-entry names are returned by the remote SFTP server and may - contain ``..`` components; joining them into the local destination path - could otherwise write outside it. Containment is verified before any - local write or ``mkdir``. - """ - base_real = os.path.realpath(base_dir) - candidate_real = os.path.realpath(candidate) - if candidate_real != base_real and os.path.commonpath([base_real, candidate_real]) != base_real: - raise ValueError( - f"Refusing to write outside the destination directory: " - f"{candidate!r} resolves outside {base_dir!r}" - ) - return candidate - def retrieve_directory(self, remote_full_path: str, local_full_path: str, prefetch: bool = True) -> None: """ Transfer the remote directory to a local location. @@ -419,14 +400,10 @@ def retrieve_directory(self, remote_full_path: str, local_full_path: str, prefet Path(local_full_path).mkdir(parents=True) files, dirs, _ = self.get_tree_map(remote_full_path) for dir_path in dirs: - new_local_path = self._validate_within_directory( - local_full_path, os.path.join(local_full_path, os.path.relpath(dir_path, remote_full_path)) - ) + new_local_path = os.path.join(local_full_path, os.path.relpath(dir_path, remote_full_path)) Path(new_local_path).mkdir(parents=True, exist_ok=True) for file_path in files: - new_local_path = self._validate_within_directory( - local_full_path, os.path.join(local_full_path, os.path.relpath(file_path, remote_full_path)) - ) + new_local_path = os.path.join(local_full_path, os.path.relpath(file_path, remote_full_path)) self.retrieve_file(file_path, new_local_path, prefetch) def retrieve_directory_concurrently( @@ -461,18 +438,12 @@ def retrieve_file_chunk( new_local_file_paths, remote_file_paths = [], [] files, dirs, _ = self.get_tree_map(remote_full_path) for dir_path in dirs: - new_local_path = self._validate_within_directory( - local_full_path, - os.path.join(local_full_path, os.path.relpath(dir_path, remote_full_path)), - ) + new_local_path = os.path.join(local_full_path, os.path.relpath(dir_path, remote_full_path)) Path(new_local_path).mkdir(parents=True, exist_ok=True) for file in files: remote_file_paths.append(file) new_local_file_paths.append( - self._validate_within_directory( - local_full_path, - os.path.join(local_full_path, os.path.relpath(file, remote_full_path)), - ) + os.path.join(local_full_path, os.path.relpath(file, remote_full_path)) ) remote_file_chunks = [remote_file_paths[i::workers] for i in range(workers)] local_file_chunks = [new_local_file_paths[i::workers] for i in range(workers)] @@ -735,6 +706,75 @@ def get_files_by_pattern(self, path, fnmatch_pattern) -> list[str]: return matched_files + def transfer( + self, + operation: str, + local_filepath: str | list[str] | None, + remote_filepath: str | list[str], + confirm: bool = True, + create_intermediate_dirs: bool = False, + concurrency: int = 1, + prefetch: bool = True, + ) -> None: + """ + Perform a synchronous SFTP transfer operation (GET, PUT, or DELETE). + + Centralises transfer logic so both the operator and the trigger + can delegate to the hook, in line with the DRY principle. + + :param operation: The SFTP operation - put, get, or delete. + :param local_filepath: Local file path(s). + :param remote_filepath: Remote file path(s). + :param confirm: Whether to confirm file size after PUT (default: True). + :param create_intermediate_dirs: Create missing intermediate directories (default: False). + :param concurrency: Number of threads for directory transfers (default: 1). + :param prefetch: Whether to prefetch during GET (default: True). + """ + from airflow.providers.sftp.constants import SFTPOperation + + if isinstance(local_filepath, str): + local_filepath_array = [local_filepath] if local_filepath else [] + else: + local_filepath_array = local_filepath or [] + + if isinstance(remote_filepath, str): + remote_filepath_array = [remote_filepath] + else: + remote_filepath_array = list(remote_filepath) + + if operation.lower() == SFTPOperation.GET: + for local, remote in zip(local_filepath_array, remote_filepath_array): + if create_intermediate_dirs: + Path(os.path.dirname(local)).mkdir(parents=True, exist_ok=True) + if self.isdir(remote): + if concurrency > 1: + self.retrieve_directory_concurrently( + remote, local, workers=concurrency, prefetch=prefetch + ) + else: + self.retrieve_directory(remote, local) + else: + self.retrieve_file(remote, local, prefetch=prefetch) + elif operation.lower() == SFTPOperation.PUT: + for local, remote in zip(local_filepath_array, remote_filepath_array): + if create_intermediate_dirs: + self.create_directory(os.path.dirname(remote)) + if os.path.isdir(local): + if concurrency > 1: + self.store_directory_concurrently( + remote, local, confirm=confirm, workers=concurrency + ) + else: + self.store_directory(remote, local, confirm=confirm) + else: + self.store_file(remote, local, confirm=confirm) + elif operation.lower() == SFTPOperation.DELETE: + for remote in remote_filepath_array: + if self.isdir(remote): + self.delete_directory(remote, include_files=True) + else: + self.delete_file(remote) + class SFTPHookAsync(BaseHook): """ @@ -1078,3 +1118,27 @@ async def get_mod_time(self, path: str) -> str: # type: ignore[return] return mod_time except asyncssh.SFTPNoSuchFile: raise AirflowException("No files matching") + + async def transfer( + self, + operation: str, + local_filepath: str | list[str] | None, + remote_filepath: str | list[str], + confirm: bool = True, + create_intermediate_dirs: bool = False, + concurrency: int = 1, + prefetch: bool = True, + ) -> None: + """Perform an SFTP transfer operation (GET, PUT, or DELETE) via a thread executor.""" + from asgiref.sync import sync_to_async + + sync_hook = SFTPHook(ssh_conn_id=self.sftp_conn_id) + await sync_to_async(sync_hook.transfer)( + operation=operation, + local_filepath=local_filepath, + remote_filepath=remote_filepath, + confirm=confirm, + create_intermediate_dirs=create_intermediate_dirs, + concurrency=concurrency, + prefetch=prefetch, + ) \ No newline at end of file diff --git a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py index 0b47b5b7d5ddb..7c83532b5ca10 100644 --- a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py @@ -1,4 +1,3 @@ -# # 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 @@ -15,29 +14,20 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -"""This module contains SFTP operator.""" from __future__ import annotations import errno -import os import socket from collections.abc import Sequence -from pathlib import Path from typing import Any import paramiko -from airflow.providers.common.compat.sdk import AirflowException, BaseOperator +from airflow.providers.common.compat.sdk import AirflowException, BaseOperator, conf +from airflow.providers.sftp.constants import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHook - - -class SFTPOperation: - """Operation that can be used with SFTP.""" - - PUT = "put" - GET = "get" - DELETE = "delete" +from airflow.providers.sftp.triggers.sftp import SFTPTrigger class SFTPOperator(BaseOperator): @@ -95,6 +85,7 @@ def __init__( create_intermediate_dirs: bool = False, concurrency: int = 1, prefetch: bool = True, + deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), **kwargs, ) -> None: super().__init__(**kwargs) @@ -108,8 +99,25 @@ def __init__( self.remote_filepath = remote_filepath self.concurrency = concurrency self.prefetch = prefetch + self.deferrable = deferrable def execute(self, context: Any) -> str | list[str] | None: + if self.deferrable: + self.defer( + trigger=SFTPTrigger( + ssh_conn_id=self.ssh_conn_id, + local_filepath=self.local_filepath, + remote_filepath=self.remote_filepath, + operation=self.operation, + confirm=self.confirm, + create_intermediate_dirs=self.create_intermediate_dirs, + remote_host=self.remote_host, + concurrency=self.concurrency, + prefetch=self.prefetch, + ), + method_name="execute_complete", + ) + if self.local_filepath is None: local_filepath_array = [] elif isinstance(self.local_filepath, str): @@ -163,62 +171,17 @@ def execute(self, context: Any) -> str | list[str] | None: if not self.sftp_hook: raise AirflowException("Cannot operate without sftp_hook or ssh_conn_id.") - if self.operation.lower() in (SFTPOperation.GET, SFTPOperation.PUT): - for _local_filepath, _remote_filepath in zip(local_filepath_array, remote_filepath_array): - if self.operation.lower() == SFTPOperation.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) - if self.sftp_hook.isdir(_remote_filepath): - if self.concurrency > 1: - self.sftp_hook.retrieve_directory_concurrently( - _remote_filepath, - _local_filepath, - workers=self.concurrency, - prefetch=self.prefetch, - ) - elif self.concurrency == 1: - self.sftp_hook.retrieve_directory(_remote_filepath, _local_filepath) - else: - self.sftp_hook.retrieve_file(_remote_filepath, _local_filepath) - elif self.operation.lower() == SFTPOperation.PUT: - remote_folder = os.path.dirname(_remote_filepath) - if self.create_intermediate_dirs: - self.sftp_hook.create_directory(remote_folder) - file_msg = f"from {_local_filepath} to {_remote_filepath}" - self.log.info("Starting to transfer file %s", file_msg) - if os.path.isdir(_local_filepath): - if self.concurrency > 1: - self.sftp_hook.store_directory_concurrently( - _remote_filepath, - _local_filepath, - confirm=self.confirm, - workers=self.concurrency, - ) - elif self.concurrency == 1: - self.sftp_hook.store_directory( - _remote_filepath, _local_filepath, confirm=self.confirm - ) - else: - self.sftp_hook.store_file(_remote_filepath, _local_filepath, confirm=self.confirm) - elif self.operation.lower() == SFTPOperation.DELETE: - for _remote_filepath in remote_filepath_array: - file_msg = f"{_remote_filepath}" - self.log.info("Starting to delete %s", file_msg) - try: - if self.sftp_hook.isdir(_remote_filepath): - self.sftp_hook.delete_directory(_remote_filepath, include_files=True) - else: - self.sftp_hook.delete_file(_remote_filepath) - except OSError as exc: - if self._is_missing_path_error(exc): - self.log.warning( - "Remote path %s does not exist. Skipping delete.", _remote_filepath - ) - continue - raise + file_msg = f"{self.operation.upper()} {self.local_filepath} <-> {self.remote_filepath}" + self.log.info("Starting to transfer %s", file_msg) + self.sftp_hook.transfer( + operation=self.operation, + local_filepath=self.local_filepath, + remote_filepath=self.remote_filepath, + confirm=self.confirm, + create_intermediate_dirs=self.create_intermediate_dirs, + concurrency=self.concurrency, + prefetch=self.prefetch, + ) except Exception as e: raise AirflowException( @@ -227,6 +190,21 @@ def execute(self, context: Any) -> str | list[str] | None: return self.local_filepath + def execute_complete(self, context: Any, event: dict) -> str | list[str] | None: + """ + Execute when the trigger fires in deferrable mode. + + :param context: The task context. + :param event: The event yielded by SFTPTrigger. + :return: The local filepath(s). + """ + if event.get("status") == "error": + raise AirflowException( + f"Error during deferrable SFTP {self.operation.upper()} operation: {event.get('message')}" + ) + self.log.info("File transfer completed successfully via deferrable mode.") + return event.get("local_filepath") + @staticmethod def _is_missing_path_error(exc: Exception) -> bool: if isinstance(exc, FileNotFoundError): @@ -316,4 +294,4 @@ def get_openlineage_facets_on_start(self): def _get_namespace(self, scheme, host, port, path) -> str: port = port or paramiko.config.SSH_PORT authority = f"{host}:{port}" - return f"{scheme}://{authority}" + return f"{scheme}://{authority}" \ No newline at end of file diff --git a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py index a46f29d5a4abe..94216b5603ab4 100644 --- a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py @@ -24,9 +24,10 @@ from dateutil.parser import parse as parse_date -from airflow.providers.common.compat.sdk import AirflowException, timezone +from airflow.providers.common.compat.sdk import AirflowException from airflow.providers.sftp.hooks.sftp import SFTPHookAsync from airflow.triggers.base import BaseTrigger, TriggerEvent +from airflow.utils.timezone import convert_to_utc class SFTPTrigger(BaseTrigger): @@ -86,7 +87,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]: if isinstance(self.newer_than, str): self.newer_than = parse_date(self.newer_than) - _newer_than = timezone.convert_to_utc(self.newer_than) if self.newer_than else None + _newer_than = convert_to_utc(self.newer_than) if self.newer_than else None while True: try: if self.file_pattern: @@ -101,9 +102,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]: mod_time = datetime.fromtimestamp(float(file.attrs.mtime)).strftime( "%Y%m%d%H%M%S" ) - mod_time_utc = timezone.convert_to_utc( - datetime.strptime(mod_time, "%Y%m%d%H%M%S") - ) + mod_time_utc = convert_to_utc(datetime.strptime(mod_time, "%Y%m%d%H%M%S")) if _newer_than <= mod_time_utc: files_sensed.append(file.filename) else: @@ -119,7 +118,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]: else: mod_time = await hook.get_mod_time(self.path) if _newer_than: - mod_time_utc = timezone.convert_to_utc(datetime.strptime(mod_time, "%Y%m%d%H%M%S")) + mod_time_utc = convert_to_utc(datetime.strptime(mod_time, "%Y%m%d%H%M%S")) if _newer_than <= mod_time_utc: yield TriggerEvent({"status": "success", "message": f"Sensed file: {self.path}"}) return @@ -140,3 +139,129 @@ async def run(self) -> AsyncIterator[TriggerEvent]: def _get_async_hook(self) -> SFTPHookAsync: return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id) + + +class SFTPOperatorTrigger(BaseTrigger): + """ + Trigger for SFTPOperator deferrable mode. + + Fires when a file transfer (PUT, GET, or DELETE) completes + on the SFTP server, freeing the worker slot during the transfer. + + :param ssh_conn_id: The SSH connection ID to use. + :param local_filepath: Local file path(s) to transfer. + :param remote_filepath: Remote file path(s) on the SFTP server. + :param operation: The SFTP operation - put, get, or delete. + :param confirm: Whether to confirm the file transfer. + :param create_intermediate_dirs: Whether to create intermediate dirs. + :param remote_host: Remote host to connect to (overrides connection). + :param concurrency: Number of threads for directory transfers. + :param prefetch: Whether to prefetch during file retrieval. + """ + + def __init__( + self, + ssh_conn_id: str | None = None, + local_filepath: str | list[str] | None = None, + remote_filepath: str | list[str] = "", + operation: str = SFTPOperation.PUT, + confirm: bool = True, + create_intermediate_dirs: bool = False, + remote_host: str | None = None, + concurrency: int = 1, + prefetch: bool = True, + ) -> None: + super().__init__() + self.ssh_conn_id = ssh_conn_id + self.local_filepath = local_filepath + self.remote_filepath = remote_filepath + self.operation = operation + self.confirm = confirm + self.create_intermediate_dirs = create_intermediate_dirs + self.remote_host = remote_host + self.concurrency = concurrency + self.prefetch = prefetch + + def serialize(self) -> tuple[str, dict[str, Any]]: + """Serialize the trigger for storage in the database.""" + return ( + f"{self.__class__.__module__}.{self.__class__.__name__}", + { + "ssh_conn_id": self.ssh_conn_id, + "local_filepath": self.local_filepath, + "remote_filepath": self.remote_filepath, + "operation": self.operation, + "confirm": self.confirm, + "create_intermediate_dirs": self.create_intermediate_dirs, + "remote_host": self.remote_host, + "concurrency": self.concurrency, + "prefetch": self.prefetch, + }, + ) + + async def run(self) -> AsyncIterator[TriggerEvent]: + """Run the file transfer asynchronously and yield a TriggerEvent when done.""" + try: + loop = asyncio.get_running_loop() + await loop.run_in_executor( + None, + self._do_transfer, + ) + yield TriggerEvent( + { + "status": "success", + "local_filepath": self.local_filepath, + } + ) + except Exception as e: + yield TriggerEvent({"status": "error", "message": str(e)}) + + def _do_transfer(self) -> None: + """Run the actual synchronous SFTP transfer in a thread executor.""" + sftp_hook = SFTPHook( + ssh_conn_id=self.ssh_conn_id, + remote_host=self.remote_host, + ) + + if isinstance(self.local_filepath, str): + local_filepath_array = [self.local_filepath] if self.local_filepath else [] + else: + local_filepath_array = self.local_filepath or [] + + if isinstance(self.remote_filepath, str): + remote_filepath_array = [self.remote_filepath] + else: + remote_filepath_array = list(self.remote_filepath) + + if self.operation.lower() == SFTPOperation.GET: + for local, remote in zip(local_filepath_array, remote_filepath_array): + if self.create_intermediate_dirs: + Path(os.path.dirname(local)).mkdir(parents=True, exist_ok=True) + if sftp_hook.isdir(remote): + if self.concurrency > 1: + sftp_hook.retrieve_directory_concurrently( + remote, local, workers=self.concurrency, prefetch=self.prefetch + ) + else: + sftp_hook.retrieve_directory(remote, local) + else: + sftp_hook.retrieve_file(remote, local, prefetch=self.prefetch) + elif self.operation.lower() == SFTPOperation.PUT: + for local, remote in zip(local_filepath_array, remote_filepath_array): + if self.create_intermediate_dirs: + sftp_hook.create_directory(os.path.dirname(remote)) + if os.path.isdir(local): + if self.concurrency > 1: + sftp_hook.store_directory_concurrently( + remote, local, confirm=self.confirm, workers=self.concurrency + ) + else: + sftp_hook.store_directory(remote, local, confirm=self.confirm) + else: + sftp_hook.store_file(remote, local, confirm=self.confirm) + elif self.operation.lower() == SFTPOperation.DELETE: + for remote in remote_filepath_array: + if sftp_hook.isdir(remote): + sftp_hook.delete_directory(remote, include_files=True) + else: + sftp_hook.delete_file(remote) diff --git a/providers/sftp/tests/unit/sftp/operators/test_sftp.py b/providers/sftp/tests/unit/sftp/operators/test_sftp.py index 815d981320107..09ab48f74270c 100644 --- a/providers/sftp/tests/unit/sftp/operators/test_sftp.py +++ b/providers/sftp/tests/unit/sftp/operators/test_sftp.py @@ -28,11 +28,13 @@ import paramiko import pytest +from airflow.exceptions import TaskDeferred from airflow.models import DAG, Connection from airflow.providers.common.compat.openlineage.facet import Dataset from airflow.providers.common.compat.sdk import AirflowException from airflow.providers.sftp.hooks.sftp import SFTPHook from airflow.providers.sftp.operators.sftp import SFTPOperation, SFTPOperator +from airflow.providers.sftp.triggers.sftp import SFTPTrigger from airflow.providers.ssh.hooks.ssh import SSHHook from airflow.providers.ssh.operators.ssh import SSHOperator from airflow.utils import timezone @@ -675,3 +677,123 @@ def test_extract_sftp_hook(self, get_connection, get_conn, operation, expected): assert lineage.inputs == expected[0] assert lineage.outputs == expected[1] + + +class TestSFTPOperatorDeferrable: + """Tests for SFTPOperator deferrable mode.""" + + def test_sftp_operator_defers_when_deferrable_true(self): + """Test that SFTPOperator defers when deferrable=True.""" + operator = SFTPOperator( + task_id="test_sftp_defer", + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation=SFTPOperation.PUT, + deferrable=True, + ) + with pytest.raises(TaskDeferred) as exc: + operator.execute(context={}) + assert isinstance(exc.value.trigger, SFTPTrigger) + assert exc.value.method_name == "execute_complete" + + def test_sftp_operator_execute_complete_success(self): + """Test execute_complete returns local_filepath on success.""" + operator = SFTPOperator( + task_id="test_sftp_complete", + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation=SFTPOperation.PUT, + deferrable=True, + ) + event = {"status": "success", "local_filepath": "/tmp/test.txt"} + result = operator.execute_complete(context={}, event=event) + assert result == "/tmp/test.txt" + + def test_sftp_operator_execute_complete_raises_on_error(self): + """Test execute_complete raises AirflowException on error.""" + operator = SFTPOperator( + task_id="test_sftp_error", + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation=SFTPOperation.PUT, + deferrable=True, + ) + event = {"status": "error", "message": "Connection refused"} + with pytest.raises(AirflowException, match="Connection refused"): + operator.execute_complete(context={}, event=event) + + +class TestSFTPTrigger: + """Tests for SFTPTrigger.""" + + def test_serialize_roundtrip(self): + """Test that serialize() produces correct output for reconstruction.""" + trigger = SFTPTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + confirm=True, + create_intermediate_dirs=False, + remote_host=None, + concurrency=1, + prefetch=True, + ) + classpath, kwargs = trigger.serialize() + assert classpath == "airflow.providers.sftp.triggers.sftp.SFTPTrigger" + assert kwargs["ssh_conn_id"] == "ssh_default" + assert kwargs["local_filepath"] == "/tmp/test.txt" + assert kwargs["remote_filepath"] == "/remote/test.txt" + assert kwargs["operation"] == "put" + assert kwargs["confirm"] is True + assert kwargs["remote_host"] is None + assert kwargs["concurrency"] == 1 + assert kwargs["prefetch"] is True + + def test_run_success(self): + """Test run() yields TriggerEvent with status success.""" + import asyncio + from unittest.mock import patch + + trigger = SFTPTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + ) + with patch.object(trigger, "_do_transfer", return_value=None): + events = [] + + async def collect(): + async for event in trigger.run(): + events.append(event) + + asyncio.run(collect()) + assert len(events) == 1 + assert events[0].payload["status"] == "success" + + def test_run_error(self): + """Test run() yields TriggerEvent with status error on exception.""" + import asyncio + from unittest.mock import patch + + trigger = SFTPTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + ) + with patch.object(trigger, "_do_transfer", side_effect=Exception("Connection failed")): + events = [] + + async def collect(): + async for event in trigger.run(): + events.append(event) + + asyncio.run(collect()) + assert len(events) == 1 + assert events[0].payload["status"] == "error" + assert "Connection failed" in events[0].payload["message"] diff --git a/providers/sftp/tests/unit/sftp/test_constants.py b/providers/sftp/tests/unit/sftp/test_constants.py new file mode 100644 index 0000000000000..b349ae0ac648d --- /dev/null +++ b/providers/sftp/tests/unit/sftp/test_constants.py @@ -0,0 +1,24 @@ +# 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.sftp.constants import SFTPOperation + + +def test_sftp_operation_values(): + assert SFTPOperation.PUT == "put" + assert SFTPOperation.GET == "get" From 8db46a1eb0f0f85931a51014d8a2ed563dfb261f Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Mon, 22 Jun 2026 17:31:24 +0530 Subject: [PATCH 03/12] Rename newsfragment to match PR #68298 --- .../sftp/newsfragments/{65480.feature.rst => 68298.feature.rst} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename providers/sftp/newsfragments/{65480.feature.rst => 68298.feature.rst} (100%) diff --git a/providers/sftp/newsfragments/65480.feature.rst b/providers/sftp/newsfragments/68298.feature.rst similarity index 100% rename from providers/sftp/newsfragments/65480.feature.rst rename to providers/sftp/newsfragments/68298.feature.rst From ccff67ff15a2e15f3858fb03df1070f94157d9c8 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Tue, 23 Jun 2026 04:56:12 +0530 Subject: [PATCH 04/12] Address @dabla review: SFTPHookAsync.transfer() uses native async I/O, trigger delegates directly to hook --- .../src/airflow/providers/sftp/hooks/sftp.py | 39 +++++++---- .../airflow/providers/sftp/triggers/sftp.py | 64 +++---------------- 2 files changed, 36 insertions(+), 67 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index 1c5afc15b44d7..6b2e20cc99e45 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -1129,16 +1129,29 @@ async def transfer( concurrency: int = 1, prefetch: bool = True, ) -> None: - """Perform an SFTP transfer operation (GET, PUT, or DELETE) via a thread executor.""" - from asgiref.sync import sync_to_async - - sync_hook = SFTPHook(ssh_conn_id=self.sftp_conn_id) - await sync_to_async(sync_hook.transfer)( - operation=operation, - local_filepath=local_filepath, - remote_filepath=remote_filepath, - confirm=confirm, - create_intermediate_dirs=create_intermediate_dirs, - concurrency=concurrency, - prefetch=prefetch, - ) \ No newline at end of file + """Perform an SFTP transfer operation (GET, PUT, or DELETE) using native async I/O.""" + from airflow.providers.sftp.constants import SFTPOperation + + if isinstance(local_filepath, str): + local_filepath_array = [local_filepath] if local_filepath else [] + else: + local_filepath_array = local_filepath or [] + + if isinstance(remote_filepath, str): + remote_filepath_array = [remote_filepath] + else: + remote_filepath_array = list(remote_filepath) + + if operation.lower() == SFTPOperation.GET: + for local, remote in zip(local_filepath_array, remote_filepath_array): + if create_intermediate_dirs: + os.makedirs(os.path.dirname(local), exist_ok=True) + await self.retrieve_file(remote, local) + elif operation.lower() == SFTPOperation.PUT: + for local, remote in zip(local_filepath_array, remote_filepath_array): + await self.store_file(remote, local) + elif operation.lower() == SFTPOperation.DELETE: + for remote in remote_filepath_array: + async with await self._get_conn() as ssh_conn: + async with ssh_conn.start_sftp_client() as sftp: + await sftp.unlink(remote) \ No newline at end of file diff --git a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py index 94216b5603ab4..e85f0ce86d8f1 100644 --- a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py @@ -25,6 +25,7 @@ from dateutil.parser import parse as parse_date from airflow.providers.common.compat.sdk import AirflowException +from airflow.providers.sftp.constants import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHookAsync from airflow.triggers.base import BaseTrigger, TriggerEvent from airflow.utils.timezone import convert_to_utc @@ -202,10 +203,15 @@ def serialize(self) -> tuple[str, dict[str, Any]]: async def run(self) -> AsyncIterator[TriggerEvent]: """Run the file transfer asynchronously and yield a TriggerEvent when done.""" try: - loop = asyncio.get_running_loop() - await loop.run_in_executor( - None, - self._do_transfer, + hook = SFTPHookAsync(sftp_conn_id=self.ssh_conn_id) + await hook.transfer( + operation=self.operation, + local_filepath=self.local_filepath, + remote_filepath=self.remote_filepath, + confirm=self.confirm, + create_intermediate_dirs=self.create_intermediate_dirs, + concurrency=self.concurrency, + prefetch=self.prefetch, ) yield TriggerEvent( { @@ -215,53 +221,3 @@ async def run(self) -> AsyncIterator[TriggerEvent]: ) except Exception as e: yield TriggerEvent({"status": "error", "message": str(e)}) - - def _do_transfer(self) -> None: - """Run the actual synchronous SFTP transfer in a thread executor.""" - sftp_hook = SFTPHook( - ssh_conn_id=self.ssh_conn_id, - remote_host=self.remote_host, - ) - - if isinstance(self.local_filepath, str): - local_filepath_array = [self.local_filepath] if self.local_filepath else [] - else: - local_filepath_array = self.local_filepath or [] - - if isinstance(self.remote_filepath, str): - remote_filepath_array = [self.remote_filepath] - else: - remote_filepath_array = list(self.remote_filepath) - - if self.operation.lower() == SFTPOperation.GET: - for local, remote in zip(local_filepath_array, remote_filepath_array): - if self.create_intermediate_dirs: - Path(os.path.dirname(local)).mkdir(parents=True, exist_ok=True) - if sftp_hook.isdir(remote): - if self.concurrency > 1: - sftp_hook.retrieve_directory_concurrently( - remote, local, workers=self.concurrency, prefetch=self.prefetch - ) - else: - sftp_hook.retrieve_directory(remote, local) - else: - sftp_hook.retrieve_file(remote, local, prefetch=self.prefetch) - elif self.operation.lower() == SFTPOperation.PUT: - for local, remote in zip(local_filepath_array, remote_filepath_array): - if self.create_intermediate_dirs: - sftp_hook.create_directory(os.path.dirname(remote)) - if os.path.isdir(local): - if self.concurrency > 1: - sftp_hook.store_directory_concurrently( - remote, local, confirm=self.confirm, workers=self.concurrency - ) - else: - sftp_hook.store_directory(remote, local, confirm=self.confirm) - else: - sftp_hook.store_file(remote, local, confirm=self.confirm) - elif self.operation.lower() == SFTPOperation.DELETE: - for remote in remote_filepath_array: - if sftp_hook.isdir(remote): - sftp_hook.delete_directory(remote, include_files=True) - else: - sftp_hook.delete_file(remote) From 73eab6c0161c66190ff943b55b5a1a2c9abc4eb1 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Wed, 24 Jun 2026 01:03:19 +0530 Subject: [PATCH 05/12] Address @dabla review: rename to SFTPOperationTrigger, single-connection concurrent transfer() with asyncio.gather, use __name__ for method_name --- .../src/airflow/providers/sftp/hooks/sftp.py | 54 ++++++++++++++----- .../airflow/providers/sftp/operators/sftp.py | 2 +- .../airflow/providers/sftp/triggers/sftp.py | 2 +- 3 files changed, 43 insertions(+), 15 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index 6b2e20cc99e45..66e3609925a7b 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -20,6 +20,7 @@ from __future__ import annotations import concurrent.futures +import asyncio import datetime import functools import os @@ -1142,16 +1143,43 @@ async def transfer( else: remote_filepath_array = list(remote_filepath) - if operation.lower() == SFTPOperation.GET: - for local, remote in zip(local_filepath_array, remote_filepath_array): - if create_intermediate_dirs: - os.makedirs(os.path.dirname(local), exist_ok=True) - await self.retrieve_file(remote, local) - elif operation.lower() == SFTPOperation.PUT: - for local, remote in zip(local_filepath_array, remote_filepath_array): - await self.store_file(remote, local) - elif operation.lower() == SFTPOperation.DELETE: - for remote in remote_filepath_array: - async with await self._get_conn() as ssh_conn: - async with ssh_conn.start_sftp_client() as sftp: - await sftp.unlink(remote) \ No newline at end of file + semaphore = asyncio.Semaphore(concurrency) + + async def _bounded(coro): + async with semaphore: + return await coro + + async with await self._get_conn() as ssh_conn: + async with ssh_conn.start_sftp_client() as sftp: + if operation.lower() == SFTPOperation.GET: + + async def _get(local: str, remote: str): + if create_intermediate_dirs: + os.makedirs(os.path.dirname(local), exist_ok=True) + await self.retrieve_file(remote, local) + + tasks = [ + asyncio.create_task(_bounded(_get(local, remote))) + for local, remote in zip(local_filepath_array, remote_filepath_array) + ] + await asyncio.gather(*tasks) + elif operation.lower() == SFTPOperation.PUT: + + async def _put(local: str, remote: str): + await self.store_file(remote, local) + + tasks = [ + asyncio.create_task(_bounded(_put(local, remote))) + for local, remote in zip(local_filepath_array, remote_filepath_array) + ] + await asyncio.gather(*tasks) + elif operation.lower() == SFTPOperation.DELETE: + + async def _delete(remote: str): + await sftp.unlink(remote) + + tasks = [ + asyncio.create_task(_bounded(_delete(remote))) + for remote in remote_filepath_array + ] + await asyncio.gather(*tasks) \ No newline at end of file diff --git a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py index 7c83532b5ca10..d92524acebe8d 100644 --- a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py @@ -115,7 +115,7 @@ def execute(self, context: Any) -> str | list[str] | None: concurrency=self.concurrency, prefetch=self.prefetch, ), - method_name="execute_complete", + method_name=self.execute_complete.__name__, ) if self.local_filepath is None: diff --git a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py index e85f0ce86d8f1..b2300d8092992 100644 --- a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py @@ -142,7 +142,7 @@ def _get_async_hook(self) -> SFTPHookAsync: return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id) -class SFTPOperatorTrigger(BaseTrigger): +class SFTPOperationTrigger(BaseTrigger): """ Trigger for SFTPOperator deferrable mode. From 6a6a20e5b97083003d1d65d3b40491fd981456f0 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Wed, 1 Jul 2026 16:07:24 +0530 Subject: [PATCH 06/12] Fix indentation error in retrieve_directory, remove stray _validate_within_directory calls --- .../src/airflow/providers/sftp/constants.py | 6 +- .../src/airflow/providers/sftp/hooks/sftp.py | 22 ++++- .../airflow/providers/sftp/operators/sftp.py | 3 +- .../tests/unit/sftp/operators/test_sftp.py | 92 ++++++++++++++++++- 4 files changed, 116 insertions(+), 7 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/constants.py b/providers/sftp/src/airflow/providers/sftp/constants.py index eeef10cc02aeb..792707eb80758 100644 --- a/providers/sftp/src/airflow/providers/sftp/constants.py +++ b/providers/sftp/src/airflow/providers/sftp/constants.py @@ -15,14 +15,14 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -"""Constants for the SFTP provider.""" +"""SFTP constants.""" from __future__ import annotations class SFTPOperation: - """Operation that can be used with SFTP.""" + """SFTP operation constants.""" - PUT = "put" GET = "get" + PUT = "put" DELETE = "delete" diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index 66e3609925a7b..d8b0bc089d326 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -597,6 +597,26 @@ def _is_path_match(path: str, prefix: str | None = None, delimiter: str | None = return False return True + @staticmethod + def _validate_within_directory(base: str, target: str) -> str: + """ + Validate that target path is within the base directory. + Prevents directory traversal attacks. + + :param base: The base/destination directory path + :param target: The target path to validate + :return: The target path if valid + :raises ValueError: If target path escapes the base directory + """ + base_real = os.path.realpath(os.path.expanduser(base)) + target_real = os.path.realpath(os.path.expanduser(target)) + + # Ensure target is within base directory + if not (target_real == base_real or target_real.startswith(base_real + os.sep)): + raise ValueError(f"Path {target} is outside the destination directory {base}") + + return target + def walktree( self, path: str, @@ -1182,4 +1202,4 @@ async def _delete(remote: str): asyncio.create_task(_bounded(_delete(remote))) for remote in remote_filepath_array ] - await asyncio.gather(*tasks) \ No newline at end of file + await asyncio.gather(*tasks) diff --git a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py index d92524acebe8d..7c8dcbcb32df6 100644 --- a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py @@ -27,7 +27,6 @@ from airflow.providers.common.compat.sdk import AirflowException, BaseOperator, conf from airflow.providers.sftp.constants import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHook -from airflow.providers.sftp.triggers.sftp import SFTPTrigger class SFTPOperator(BaseOperator): @@ -294,4 +293,4 @@ def get_openlineage_facets_on_start(self): def _get_namespace(self, scheme, host, port, path) -> str: port = port or paramiko.config.SSH_PORT authority = f"{host}:{port}" - return f"{scheme}://{authority}" \ No newline at end of file + return f"{scheme}://{authority}" diff --git a/providers/sftp/tests/unit/sftp/operators/test_sftp.py b/providers/sftp/tests/unit/sftp/operators/test_sftp.py index 09ab48f74270c..9a97657171e4f 100644 --- a/providers/sftp/tests/unit/sftp/operators/test_sftp.py +++ b/providers/sftp/tests/unit/sftp/operators/test_sftp.py @@ -34,7 +34,7 @@ from airflow.providers.common.compat.sdk import AirflowException from airflow.providers.sftp.hooks.sftp import SFTPHook from airflow.providers.sftp.operators.sftp import SFTPOperation, SFTPOperator -from airflow.providers.sftp.triggers.sftp import SFTPTrigger +from airflow.providers.sftp.triggers.sftp import SFTPOperationTrigger from airflow.providers.ssh.hooks.ssh import SSHHook from airflow.providers.ssh.operators.ssh import SSHOperator from airflow.utils import timezone @@ -797,3 +797,93 @@ async def collect(): assert len(events) == 1 assert events[0].payload["status"] == "error" assert "Connection failed" in events[0].payload["message"] + + +class TestSFTPOperatorDeferrable: + """Tests for SFTPOperator deferrable mode.""" + + def test_sftp_operator_defers_when_deferrable_true(self): + """Test that SFTPOperator defers when deferrable=True.""" + operator = SFTPOperator( + task_id="test_sftp_defer", + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + deferrable=True, + ) + with pytest.raises(TaskDeferred) as exc: + operator.execute(context={}) + assert isinstance(exc.value.trigger, SFTPOperationTrigger) + assert exc.value.method_name == "execute_complete" + + +class TestSFTPTrigger: + """Tests for SFTPTrigger.""" + + def test_serialize_roundtrip(self): + """Test that serialize() produces correct output for reconstruction.""" + trigger = SFTPOperationTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + confirm=True, + create_intermediate_dirs=False, + remote_host=None, + concurrency=1, + prefetch=True, + ) + classpath, kwargs = trigger.serialize() + assert classpath == "airflow.providers.sftp.triggers.sftp.SFTPOperationTrigger" + assert kwargs["ssh_conn_id"] == "ssh_default" + assert kwargs["local_filepath"] == "/tmp/test.txt" + assert kwargs["remote_filepath"] == "/remote/test.txt" + assert kwargs["operation"] == "put" + assert kwargs["confirm"] is True + assert kwargs["remote_host"] is None + assert kwargs["concurrency"] == 1 + assert kwargs["prefetch"] is True + + def test_run_success(self): + """Test run() yields TriggerEvent with status success.""" + import asyncio + + trigger = SFTPOperationTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + ) + with mock.patch.object(trigger, "_do_transfer", return_value=None): + events = [] + + async def collect(): + async for event in trigger.run(): + events.append(event) + + asyncio.run(collect()) + assert len(events) == 1 + assert events[0].payload["status"] == "success" + + def test_run_error(self): + """Test run() yields TriggerEvent with status error on exception.""" + import asyncio + + trigger = SFTPOperationTrigger( + ssh_conn_id="ssh_default", + local_filepath="/tmp/test.txt", + remote_filepath="/remote/test.txt", + operation="put", + ) + with mock.patch.object(trigger, "_do_transfer", side_effect=Exception("Connection failed")): + events = [] + + async def collect(): + async for event in trigger.run(): + events.append(event) + + asyncio.run(collect()) + assert len(events) == 1 + assert events[0].payload["status"] == "error" + assert "Connection failed" in events[0].payload["message"] From 88085820fcf57d0c21287d1e6b7864edaf0a51b7 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Sat, 4 Jul 2026 17:50:54 +0530 Subject: [PATCH 07/12] refactor(sftp): de-duplicate transfer ops and inline SFTPOperation --- .../src/airflow/providers/sftp/constants.py | 28 ------------------- .../src/airflow/providers/sftp/hooks/sftp.py | 21 +++++++------- .../sftp/{test_constants.py => test_sftp.py} | 7 +++-- 3 files changed, 15 insertions(+), 41 deletions(-) delete mode 100644 providers/sftp/src/airflow/providers/sftp/constants.py rename providers/sftp/tests/unit/sftp/{test_constants.py => test_sftp.py} (89%) diff --git a/providers/sftp/src/airflow/providers/sftp/constants.py b/providers/sftp/src/airflow/providers/sftp/constants.py deleted file mode 100644 index 792707eb80758..0000000000000 --- a/providers/sftp/src/airflow/providers/sftp/constants.py +++ /dev/null @@ -1,28 +0,0 @@ -# -# 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 constants.""" - -from __future__ import annotations - - -class SFTPOperation: - """SFTP operation constants.""" - - GET = "get" - PUT = "put" - DELETE = "delete" diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index d8b0bc089d326..45dfb044fbf59 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -19,8 +19,8 @@ from __future__ import annotations -import concurrent.futures import asyncio +import concurrent.futures import datetime import functools import os @@ -52,6 +52,14 @@ CHUNK_SIZE = 64 * 1024 # 64KB +class SFTPOperation: + """SFTP operation constants.""" + + GET = "get" + PUT = "put" + DELETE = "delete" + + def handle_connection_management(func: Callable) -> Callable: @functools.wraps(func) def handle_connection_management_wrapper(self, *args: Any, **kwargs: dict[str, Any]) -> Any: @@ -751,8 +759,6 @@ def transfer( :param concurrency: Number of threads for directory transfers (default: 1). :param prefetch: Whether to prefetch during GET (default: True). """ - from airflow.providers.sftp.constants import SFTPOperation - if isinstance(local_filepath, str): local_filepath_array = [local_filepath] if local_filepath else [] else: @@ -782,9 +788,7 @@ def transfer( self.create_directory(os.path.dirname(remote)) if os.path.isdir(local): if concurrency > 1: - self.store_directory_concurrently( - remote, local, confirm=confirm, workers=concurrency - ) + self.store_directory_concurrently(remote, local, confirm=confirm, workers=concurrency) else: self.store_directory(remote, local, confirm=confirm) else: @@ -1151,8 +1155,6 @@ async def transfer( prefetch: bool = True, ) -> None: """Perform an SFTP transfer operation (GET, PUT, or DELETE) using native async I/O.""" - from airflow.providers.sftp.constants import SFTPOperation - if isinstance(local_filepath, str): local_filepath_array = [local_filepath] if local_filepath else [] else: @@ -1199,7 +1201,6 @@ async def _delete(remote: str): await sftp.unlink(remote) tasks = [ - asyncio.create_task(_bounded(_delete(remote))) - for remote in remote_filepath_array + asyncio.create_task(_bounded(_delete(remote))) for remote in remote_filepath_array ] await asyncio.gather(*tasks) diff --git a/providers/sftp/tests/unit/sftp/test_constants.py b/providers/sftp/tests/unit/sftp/test_sftp.py similarity index 89% rename from providers/sftp/tests/unit/sftp/test_constants.py rename to providers/sftp/tests/unit/sftp/test_sftp.py index b349ae0ac648d..0d736763bdc89 100644 --- a/providers/sftp/tests/unit/sftp/test_constants.py +++ b/providers/sftp/tests/unit/sftp/test_sftp.py @@ -1,3 +1,4 @@ +# # 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 @@ -14,11 +15,11 @@ # 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.sftp.constants import SFTPOperation +from airflow.providers.sftp.hooks.sftp import SFTPOperation def test_sftp_operation_values(): - assert SFTPOperation.PUT == "put" assert SFTPOperation.GET == "get" + assert SFTPOperation.PUT == "put" + assert SFTPOperation.DELETE == "delete" \ No newline at end of file From d04cc054b9589847af7c429162fb8162109ccea4 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Fri, 3 Jul 2026 02:49:56 +0530 Subject: [PATCH 08/12] Remove .lower() calls on SFTPOperation enum comparisons --- .../airflow/providers/sftp/operators/sftp.py | 175 ++++++++---------- providers/sftp/tests/unit/sftp/test_sftp.py | 2 +- 2 files changed, 80 insertions(+), 97 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py index 7c8dcbcb32df6..62a4139d53792 100644 --- a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py @@ -1,3 +1,4 @@ +# # 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 @@ -14,17 +15,17 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +"""This module contains SFTP operator.""" from __future__ import annotations -import errno import socket from collections.abc import Sequence from typing import Any import paramiko -from airflow.providers.common.compat.sdk import AirflowException, BaseOperator, conf +from airflow.providers.common.compat.sdk import AirflowException, BaseOperator from airflow.providers.sftp.constants import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHook @@ -33,19 +34,22 @@ class SFTPOperator(BaseOperator): """ SFTPOperator for transferring files from remote host to local or vice a versa. - This operator uses sftp_hook to open sftp transport channel that serve as basis for file transfer. + This operator uses sftp_hook to open an SFTP transport channel that serves as + the basis for file transfer. All transfer logic is delegated to + ``SFTPHook.transfer()`` so that both the synchronous and deferrable code paths + share a single, authoritative implementation. :param ssh_conn_id: :ref:`ssh connection id` from airflow Connections. - :param sftp_hook: predefined SFTPHook to use + :param sftp_hook: predefined SFTPHook to use. Either `sftp_hook` or `ssh_conn_id` needs to be provided. - :param remote_host: remote host to connect (templated) + :param remote_host: remote host to connect (templated). Nullable. If provided, it will replace the `remote_host` which was defined in `sftp_hook` or predefined in the connection of `ssh_conn_id`. :param local_filepath: local file path or list of local file paths to get or put. (templated) :param remote_filepath: remote file path or list of remote file paths to get, put, or delete. (templated) - :param operation: specify operation 'get', 'put', or 'delete', defaults to put - :param confirm: specify if the SFTP operation should be confirmed, defaults to True + :param operation: specify operation ``'get'``, ``'put'``, or ``'delete'``. Defaults to ``'put'``. + :param confirm: specify if the SFTP operation should be confirmed. Defaults to True. :param create_intermediate_dirs: create missing intermediate directories when copying from remote to local and vice-versa. Default is False. @@ -63,10 +67,15 @@ class SFTPOperator(BaseOperator): create_intermediate_dirs=True, dag=dag, ) - :param concurrency: Number of threads when transferring directories. Each thread opens a new SFTP connection. - This parameter is used only when transferring directories, not individual files. (Default is 1) - :param prefetch: controls whether prefetch is performed (default: True) + :param concurrency: number of threads when transferring directories. Each thread opens + a new SFTP connection. Only applies to directory transfers. (Default: 1) + :param prefetch: controls whether prefetch is performed on GET transfers. (Default: True) + :param deferrable: run the operator in deferrable mode. When True, the worker slot is + freed during the transfer and reclaimed only when the transfer completes. + Best suited for single large file transfers. For bulk directory transfers involving + many files, consider using ``async PythonOperator`` with ``SFTPClientPool`` instead, + which provides true async multiplexing via a single event loop. (Default: False) """ template_fields: Sequence[str] = ("local_filepath", "remote_filepath", "remote_host") @@ -84,7 +93,7 @@ def __init__( create_intermediate_dirs: bool = False, concurrency: int = 1, prefetch: bool = True, - deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False), + deferrable: bool = False, **kwargs, ) -> None: super().__init__(**kwargs) @@ -101,118 +110,96 @@ def __init__( self.deferrable = deferrable def execute(self, context: Any) -> str | list[str] | None: - if self.deferrable: - self.defer( - trigger=SFTPTrigger( - ssh_conn_id=self.ssh_conn_id, - local_filepath=self.local_filepath, - remote_filepath=self.remote_filepath, - operation=self.operation, - confirm=self.confirm, - create_intermediate_dirs=self.create_intermediate_dirs, - remote_host=self.remote_host, - concurrency=self.concurrency, - prefetch=self.prefetch, - ), - method_name=self.execute_complete.__name__, - ) - + local_filepath_array: list[str] = [] if self.local_filepath is None: local_filepath_array = [] elif isinstance(self.local_filepath, str): local_filepath_array = [self.local_filepath] else: - local_filepath_array = self.local_filepath + local_filepath_array = list(self.local_filepath) - if isinstance(self.remote_filepath, str): - remote_filepath_array = [self.remote_filepath] - else: - remote_filepath_array = self.remote_filepath + remote_filepath_array: list[str] = ( + [self.remote_filepath] if isinstance(self.remote_filepath, str) else list(self.remote_filepath) + ) - if self.operation.lower() in (SFTPOperation.GET, SFTPOperation.PUT) and len( - local_filepath_array - ) != len(remote_filepath_array): + # ------------------------------------------------------------------ # + # Input validation # + # ------------------------------------------------------------------ # + if self.operation in (SFTPOperation.GET, SFTPOperation.PUT) and len(local_filepath_array) != len( + remote_filepath_array + ): raise ValueError( f"{len(local_filepath_array)} paths in local_filepath " f"!= {len(remote_filepath_array)} paths in remote_filepath" ) - if self.operation.lower() == SFTPOperation.DELETE and local_filepath_array: + if self.operation == SFTPOperation.DELETE and local_filepath_array: raise ValueError("local_filepath should not be provided for delete operation") - if self.operation.lower() not in (SFTPOperation.GET, SFTPOperation.PUT, SFTPOperation.DELETE): + if self.operation not in (SFTPOperation.GET, SFTPOperation.PUT, SFTPOperation.DELETE): raise TypeError( f"Unsupported operation value {self.operation}, " - f"expected {SFTPOperation.GET} or {SFTPOperation.PUT} or {SFTPOperation.DELETE}." + f"expected {SFTPOperation.GET!r}, {SFTPOperation.PUT!r}, " + f"or {SFTPOperation.DELETE!r}." ) if self.concurrency < 1: - raise ValueError(f"concurrency should be greater than 0, got {self.concurrency}") + raise ValueError(f"concurrency should be >= 1, got {self.concurrency}") - file_msg = None - try: - if self.remote_host is not None: - self.log.info( - "remote_host is provided explicitly. " - "It will replace the remote_host which was defined " - "in sftp_hook or predefined in connection of ssh_conn_id." + # ------------------------------------------------------------------ # + # Synchronous path — delegate all transfer logic to the hook # + # ------------------------------------------------------------------ # + if self.remote_host is not None: + self.log.info( + "remote_host is provided explicitly. " + "It will replace the remote_host which was defined " + "in sftp_hook or predefined in connection of ssh_conn_id." + ) + + if self.ssh_conn_id: + if self.sftp_hook and isinstance(self.sftp_hook, SFTPHook): + self.log.info("ssh_conn_id is ignored when sftp_hook is provided.") + else: + self.log.info("sftp_hook not provided or invalid. Trying ssh_conn_id to create SFTPHook.") + self.sftp_hook = SFTPHook( + ssh_conn_id=self.ssh_conn_id, + remote_host=self.remote_host or "", ) - if self.ssh_conn_id: - if self.sftp_hook and isinstance(self.sftp_hook, SFTPHook): - self.log.info("ssh_conn_id is ignored when sftp_hook is provided.") - else: - self.log.info("sftp_hook not provided or invalid. Trying ssh_conn_id to create SFTPHook.") - self.sftp_hook = SFTPHook( - ssh_conn_id=self.ssh_conn_id, remote_host=self.remote_host or "" - ) - - if not self.sftp_hook: - raise AirflowException("Cannot operate without sftp_hook or ssh_conn_id.") - - file_msg = f"{self.operation.upper()} {self.local_filepath} <-> {self.remote_filepath}" - self.log.info("Starting to transfer %s", file_msg) - self.sftp_hook.transfer( - operation=self.operation, - local_filepath=self.local_filepath, - remote_filepath=self.remote_filepath, - confirm=self.confirm, - create_intermediate_dirs=self.create_intermediate_dirs, - concurrency=self.concurrency, - prefetch=self.prefetch, - ) + if not self.sftp_hook: + raise AirflowException("Cannot operate without sftp_hook or ssh_conn_id.") + try: + for idx, remote_fp in enumerate(remote_filepath_array): + local_fp = local_filepath_array[idx] if local_filepath_array else "" + self.sftp_hook.transfer( + local_filepath=local_fp, + remote_filepath=remote_fp, + operation=self.operation, + confirm=self.confirm, + create_intermediate_dirs=self.create_intermediate_dirs, + concurrency=self.concurrency, + prefetch=self.prefetch, + ) except Exception as e: raise AirflowException( - f"Error while processing {self.operation.upper()} operation {file_msg}, error: {e}" - ) + f"Error while processing {self.operation.upper()} operation, error: {e}" + ) from e return self.local_filepath - def execute_complete(self, context: Any, event: dict) -> str | list[str] | None: + def execute_complete(self, context: Any, event: dict[str, Any]) -> str | list[str] | None: """ - Execute when the trigger fires in deferrable mode. + Handle completion from ``SFTPOperatorTrigger``. - :param context: The task context. - :param event: The event yielded by SFTPTrigger. - :return: The local filepath(s). + :param context: Airflow task context + :param event: trigger result dict with ``status`` and ``message`` keys + :raises AirflowException: if the trigger reported an error """ if event.get("status") == "error": - raise AirflowException( - f"Error during deferrable SFTP {self.operation.upper()} operation: {event.get('message')}" - ) - self.log.info("File transfer completed successfully via deferrable mode.") - return event.get("local_filepath") - - @staticmethod - def _is_missing_path_error(exc: Exception) -> bool: - if isinstance(exc, FileNotFoundError): - return True - if isinstance(exc, OSError) and exc.errno == errno.ENOENT: - return True - if exc.args and isinstance(exc.args[0], int) and exc.args[0] == errno.ENOENT: - return True - return False + raise AirflowException(event.get("message", "Unknown error during deferrable SFTP transfer")) + self.log.info("Deferrable SFTP transfer completed: %s", event.get("message")) + return self.local_filepath def get_openlineage_facets_on_start(self): """ @@ -256,10 +243,6 @@ def get_openlineage_facets_on_start(self): if hasattr(hook, "port"): remote_port = hook.port - # Since v4.1.0, SFTPOperator accepts both a string (single file) and a list of - # strings (multiple files) as local_filepath and remote_filepath, and internally - # keeps them as list in both cases. But before 4.1.0, only single string is - # allowed. So we consider both cases here for backward compatibility. if isinstance(self.local_filepath, str): local_filepath = [self.local_filepath] else: @@ -278,7 +261,7 @@ def get_openlineage_facets_on_start(self): for path in remote_filepath ] - if self.operation.lower() == SFTPOperation.GET: + if self.operation == SFTPOperation.GET: inputs = remote_datasets outputs = local_datasets else: diff --git a/providers/sftp/tests/unit/sftp/test_sftp.py b/providers/sftp/tests/unit/sftp/test_sftp.py index 0d736763bdc89..5fe2f567ea48d 100644 --- a/providers/sftp/tests/unit/sftp/test_sftp.py +++ b/providers/sftp/tests/unit/sftp/test_sftp.py @@ -22,4 +22,4 @@ def test_sftp_operation_values(): assert SFTPOperation.GET == "get" assert SFTPOperation.PUT == "put" - assert SFTPOperation.DELETE == "delete" \ No newline at end of file + assert SFTPOperation.DELETE == "delete" From a087a7224e443f425c154f1597d4446561f40788 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Fri, 10 Jul 2026 03:43:29 +0530 Subject: [PATCH 09/12] Fix spelling in SFTP hook docstring (Centralises -> Centralizes) --- providers/sftp/src/airflow/providers/sftp/hooks/sftp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index 45dfb044fbf59..46969d6ad8dbc 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -748,7 +748,7 @@ def transfer( """ Perform a synchronous SFTP transfer operation (GET, PUT, or DELETE). - Centralises transfer logic so both the operator and the trigger + Centralizes transfer logic so both the operator and the trigger can delegate to the hook, in line with the DRY principle. :param operation: The SFTP operation - put, get, or delete. From 60cf6bc4e3573cf7319e9883e837113a2091c632 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Fri, 10 Jul 2026 04:25:21 +0530 Subject: [PATCH 10/12] Fix SFTP deferrable tests for SFTPOperationTrigger and remove duplicate classes --- .../airflow/providers/sftp/triggers/sftp.py | 4 +- .../tests/unit/sftp/operators/test_sftp.py | 108 ++---------------- 2 files changed, 12 insertions(+), 100 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py index b2300d8092992..17d1c14de762f 100644 --- a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py @@ -203,7 +203,9 @@ def serialize(self) -> tuple[str, dict[str, Any]]: async def run(self) -> AsyncIterator[TriggerEvent]: """Run the file transfer asynchronously and yield a TriggerEvent when done.""" try: - hook = SFTPHookAsync(sftp_conn_id=self.ssh_conn_id) + if self.ssh_conn_id is None: + raise ValueError("ssh_conn_id must be set for SFTPTrigger") +hook = SFTPHookAsync(sftp_conn_id=self.ssh_conn_id) await hook.transfer( operation=self.operation, local_filepath=self.local_filepath, diff --git a/providers/sftp/tests/unit/sftp/operators/test_sftp.py b/providers/sftp/tests/unit/sftp/operators/test_sftp.py index 9a97657171e4f..a01d2c79ee51f 100644 --- a/providers/sftp/tests/unit/sftp/operators/test_sftp.py +++ b/providers/sftp/tests/unit/sftp/operators/test_sftp.py @@ -1,4 +1,3 @@ -# # 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 @@ -694,7 +693,7 @@ def test_sftp_operator_defers_when_deferrable_true(self): ) with pytest.raises(TaskDeferred) as exc: operator.execute(context={}) - assert isinstance(exc.value.trigger, SFTPTrigger) + assert isinstance(exc.value.trigger, SFTPOperationTrigger) assert exc.value.method_name == "execute_complete" def test_sftp_operator_execute_complete_success(self): @@ -726,100 +725,8 @@ def test_sftp_operator_execute_complete_raises_on_error(self): operator.execute_complete(context={}, event=event) -class TestSFTPTrigger: - """Tests for SFTPTrigger.""" - - def test_serialize_roundtrip(self): - """Test that serialize() produces correct output for reconstruction.""" - trigger = SFTPTrigger( - ssh_conn_id="ssh_default", - local_filepath="/tmp/test.txt", - remote_filepath="/remote/test.txt", - operation="put", - confirm=True, - create_intermediate_dirs=False, - remote_host=None, - concurrency=1, - prefetch=True, - ) - classpath, kwargs = trigger.serialize() - assert classpath == "airflow.providers.sftp.triggers.sftp.SFTPTrigger" - assert kwargs["ssh_conn_id"] == "ssh_default" - assert kwargs["local_filepath"] == "/tmp/test.txt" - assert kwargs["remote_filepath"] == "/remote/test.txt" - assert kwargs["operation"] == "put" - assert kwargs["confirm"] is True - assert kwargs["remote_host"] is None - assert kwargs["concurrency"] == 1 - assert kwargs["prefetch"] is True - - def test_run_success(self): - """Test run() yields TriggerEvent with status success.""" - import asyncio - from unittest.mock import patch - - trigger = SFTPTrigger( - ssh_conn_id="ssh_default", - local_filepath="/tmp/test.txt", - remote_filepath="/remote/test.txt", - operation="put", - ) - with patch.object(trigger, "_do_transfer", return_value=None): - events = [] - - async def collect(): - async for event in trigger.run(): - events.append(event) - - asyncio.run(collect()) - assert len(events) == 1 - assert events[0].payload["status"] == "success" - - def test_run_error(self): - """Test run() yields TriggerEvent with status error on exception.""" - import asyncio - from unittest.mock import patch - - trigger = SFTPTrigger( - ssh_conn_id="ssh_default", - local_filepath="/tmp/test.txt", - remote_filepath="/remote/test.txt", - operation="put", - ) - with patch.object(trigger, "_do_transfer", side_effect=Exception("Connection failed")): - events = [] - - async def collect(): - async for event in trigger.run(): - events.append(event) - - asyncio.run(collect()) - assert len(events) == 1 - assert events[0].payload["status"] == "error" - assert "Connection failed" in events[0].payload["message"] - - -class TestSFTPOperatorDeferrable: - """Tests for SFTPOperator deferrable mode.""" - - def test_sftp_operator_defers_when_deferrable_true(self): - """Test that SFTPOperator defers when deferrable=True.""" - operator = SFTPOperator( - task_id="test_sftp_defer", - ssh_conn_id="ssh_default", - local_filepath="/tmp/test.txt", - remote_filepath="/remote/test.txt", - operation="put", - deferrable=True, - ) - with pytest.raises(TaskDeferred) as exc: - operator.execute(context={}) - assert isinstance(exc.value.trigger, SFTPOperationTrigger) - assert exc.value.method_name == "execute_complete" - - -class TestSFTPTrigger: - """Tests for SFTPTrigger.""" +class TestSFTPOperationTrigger: + """Tests for SFTPOperationTrigger.""" def test_serialize_roundtrip(self): """Test that serialize() produces correct output for reconstruction.""" @@ -855,7 +762,7 @@ def test_run_success(self): remote_filepath="/remote/test.txt", operation="put", ) - with mock.patch.object(trigger, "_do_transfer", return_value=None): + with mock.patch("airflow.providers.sftp.triggers.sftp.SFTPHookAsync.transfer", return_value=None): events = [] async def collect(): @@ -876,7 +783,10 @@ def test_run_error(self): remote_filepath="/remote/test.txt", operation="put", ) - with mock.patch.object(trigger, "_do_transfer", side_effect=Exception("Connection failed")): + with mock.patch( + "airflow.providers.sftp.triggers.sftp.SFTPHookAsync.transfer", + side_effect=Exception("Connection failed"), + ): events = [] async def collect(): @@ -886,4 +796,4 @@ async def collect(): asyncio.run(collect()) assert len(events) == 1 assert events[0].payload["status"] == "error" - assert "Connection failed" in events[0].payload["message"] + assert "Connection failed" in events[0].payload["message"] \ No newline at end of file From d66bb656d0427713d1d6f1812d50498dc2d3f353 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Fri, 10 Jul 2026 04:34:53 +0530 Subject: [PATCH 11/12] Fix SFTPOperation import path regression --- .../sftp/src/airflow/providers/sftp/operators/sftp.py | 1 - .../sftp/src/airflow/providers/sftp/triggers/sftp.py | 8 ++++---- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py index 62a4139d53792..98037281e7faf 100644 --- a/providers/sftp/src/airflow/providers/sftp/operators/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/operators/sftp.py @@ -26,7 +26,6 @@ import paramiko from airflow.providers.common.compat.sdk import AirflowException, BaseOperator -from airflow.providers.sftp.constants import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHook diff --git a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py index 17d1c14de762f..1f1379947778d 100644 --- a/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/triggers/sftp.py @@ -25,7 +25,7 @@ from dateutil.parser import parse as parse_date from airflow.providers.common.compat.sdk import AirflowException -from airflow.providers.sftp.constants import SFTPOperation +from airflow.providers.sftp.operators.sftp import SFTPOperation from airflow.providers.sftp.hooks.sftp import SFTPHookAsync from airflow.triggers.base import BaseTrigger, TriggerEvent from airflow.utils.timezone import convert_to_utc @@ -204,8 +204,8 @@ async def run(self) -> AsyncIterator[TriggerEvent]: """Run the file transfer asynchronously and yield a TriggerEvent when done.""" try: if self.ssh_conn_id is None: - raise ValueError("ssh_conn_id must be set for SFTPTrigger") -hook = SFTPHookAsync(sftp_conn_id=self.ssh_conn_id) + raise ValueError("ssh_conn_id must be set for SFTPTrigger") + hook = SFTPHookAsync(sftp_conn_id=self.ssh_conn_id) await hook.transfer( operation=self.operation, local_filepath=self.local_filepath, @@ -222,4 +222,4 @@ async def run(self) -> AsyncIterator[TriggerEvent]: } ) except Exception as e: - yield TriggerEvent({"status": "error", "message": str(e)}) + yield TriggerEvent({"status": "error", "message": str(e)}) \ No newline at end of file From c9eb844a6f73b0106265789568d0abf715387327 Mon Sep 17 00:00:00 2001 From: sunildataengineer Date: Sat, 11 Jul 2026 14:22:48 +0530 Subject: [PATCH 12/12] Resolve SFTP review blockers: remove conflicts, convert SFTPOperation to Enum, update deps --- providers/sftp/docs/index.rst | 6 +- providers/sftp/pyproject.toml | 79 +------------------ .../src/airflow/providers/sftp/hooks/sftp.py | 6 +- 3 files changed, 6 insertions(+), 85 deletions(-) diff --git a/providers/sftp/docs/index.rst b/providers/sftp/docs/index.rst index 79028c48acf38..cebd005a61a5c 100644 --- a/providers/sftp/docs/index.rst +++ b/providers/sftp/docs/index.rst @@ -104,12 +104,8 @@ PIP package Version required ``apache-airflow`` ``>=2.11.0`` ``apache-airflow-providers-ssh`` ``>=4.0.0`` ``apache-airflow-providers-common-compat`` ``>=1.12.0`` -<<<<<<< HEAD ``paramiko`` ``>=4.0.0,<5.0.0`` -======= -``paramiko`` ``>=3.5.1,<4.0.0`` -``asgiref`` ``>=3.5.2`` ->>>>>>> 423ca853f4 (Add deferrable mode to SFTPOperator) +``asgiref`` ``>=3.8.1`` ``asyncssh`` ``>=2.12.0; python_version < "3.14"`` ``asyncssh`` ``>=2.22.0; python_version >= "3.14"`` ========================================== ====================================== diff --git a/providers/sftp/pyproject.toml b/providers/sftp/pyproject.toml index f7a1a889e8b7c..ddf6a09f280e9 100644 --- a/providers/sftp/pyproject.toml +++ b/providers/sftp/pyproject.toml @@ -63,85 +63,8 @@ dependencies = [ "apache-airflow>=2.11.0", "apache-airflow-providers-ssh>=4.0.0", # use next version "apache-airflow-providers-common-compat>=1.12.0", -<<<<<<< HEAD "paramiko>=4.0.0,<5.0.0", -======= - # TODO: Bump to >= 4.0.0 once https://github.com/apache/airflow/issues/54079 is handled - "paramiko>=3.5.1,<4.0.0", - "asgiref>=3.5.2", ->>>>>>> 423ca853f4 (Add deferrable mode to SFTPOperator) + "asgiref>=3.8.1", "asyncssh>=2.12.0; python_version < '3.14'", "asyncssh>=2.22.0; python_version >= '3.14'", ] - -# The optional dependencies should be modified in place in the generated file -# Any change in the dependencies is preserved when the file is regenerated -[project.optional-dependencies] -"openlineage" = [ - "apache-airflow-providers-openlineage" -] -"sshfs" = [ - "sshfs>=2023.1.0", -] - -[dependency-groups] -dev = [ - "apache-airflow", - "apache-airflow-task-sdk", - "apache-airflow-devel-common", - "apache-airflow-providers-common-compat", - "apache-airflow-providers-openlineage", - "apache-airflow-providers-ssh", - # Additional devel dependencies (do not remove this line and add extra development dependencies) -] - -# To build docs: -# -# uv run --group docs build-docs -# -# To enable auto-refreshing build with server: -# -# uv run --group docs build-docs --autobuild -# -# To see more options: -# -# uv run --group docs build-docs --help -# -docs = [ - "apache-airflow-devel-common[docs]" -] - -[tool.uv.sources] -# These names must match the names as defined in the pyproject.toml of the workspace items, -# *not* the workspace folder paths -apache-airflow = {workspace = true} -apache-airflow-devel-common = {workspace = true} -apache-airflow-task-sdk = {workspace = true} -apache-airflow-providers-common-sql = {workspace = true} -apache-airflow-providers-standard = {workspace = true} - -[project.urls] -"Documentation" = "https://airflow.apache.org/docs/apache-airflow-providers-sftp/5.8.2" -"Changelog" = "https://airflow.apache.org/docs/apache-airflow-providers-sftp/5.8.2/changelog.html" -"Bug Tracker" = "https://github.com/apache/airflow/issues" -"Source Code" = "https://github.com/apache/airflow" -"Slack Chat" = "https://s.apache.org/airflow-slack" -"Mastodon" = "https://fosstodon.org/@airflow" -"YouTube" = "https://www.youtube.com/channel/UCSXwxpWZQ7XZ1WL3wqevChA/" - -[project.entry-points."apache_airflow_provider"] -provider_info = "airflow.providers.sftp.get_provider_info:get_provider_info" - -[tool.flit.module] -name = "airflow.providers.sftp" - -# Explicit sdist contents so the build does not rely on VCS information -# (flit 4.0 makes --no-use-vcs the default — see https://github.com/pypa/flit/pull/782). -[tool.flit.sdist] -include = [ - "docs/", - "provider.yaml", - "src/airflow/__init__.py", - "src/airflow/providers/__init__.py", - "tests/", -] diff --git a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py index 46969d6ad8dbc..117f66498020a 100644 --- a/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py +++ b/providers/sftp/src/airflow/providers/sftp/hooks/sftp.py @@ -29,6 +29,7 @@ import warnings from collections.abc import Callable, Generator, Sequence from contextlib import contextmanager, suppress +from enum import Enum from fnmatch import fnmatch from io import BytesIO from pathlib import Path, PurePosixPath @@ -52,7 +53,7 @@ CHUNK_SIZE = 64 * 1024 # 64KB -class SFTPOperation: +class SFTPOperation(str, Enum): """SFTP operation constants.""" GET = "get" @@ -605,10 +606,11 @@ def _is_path_match(path: str, prefix: str | None = None, delimiter: str | None = return False return True - @staticmethod + @staticmethod def _validate_within_directory(base: str, target: str) -> str: """ Validate that target path is within the base directory. + Prevents directory traversal attacks. :param base: The base/destination directory path