bugraoz93 commented on code in PR #68298:
URL: https://github.com/apache/airflow/pull/68298#discussion_r3508258484
##########
providers/sftp/src/airflow/providers/sftp/hooks/sftp.py:
##########
@@ -1069,3 +1130,67 @@ 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,
Review Comment:
I would expect this to be enum with above command to match rather string
comparison, we can use as type and compare directly without even values here.
Maybe to cast, we should create enum to pass the `operation` on caller method
with constructor of enum with values only `enum_class[str]`
https://docs.python.org/3/library/enum.html#enum.EnumType.__getitem__
##########
providers/sftp/pyproject.toml:
##########
@@ -65,6 +65,7 @@ dependencies = [
"apache-airflow-providers-common-compat>=1.12.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",
Review Comment:
This should be latest. Do we have any constraint using release from 2022?
https://pypi.org/project/asgiref/3.11.1/
##########
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.
+"""SFTP constants."""
+
+from __future__ import annotations
+
+
+class SFTPOperation:
Review Comment:
This class looked Enum to me :)
##########
providers/sftp/src/airflow/providers/sftp/triggers/sftp.py:
##########
@@ -140,3 +140,84 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
def _get_async_hook(self) -> SFTPHookAsync:
return SFTPHookAsync(sftp_conn_id=self.sftp_conn_id)
+
+
+class SFTPOperationTrigger(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,
Review Comment:
Enum, similar to above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
