Add SSH/SFTP support for ObjectStoragePath

providers/sftp/docs/filesystems/index.rst

@@ -0,0 +1,26 @@
+ .. 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.
+
+Filesystems
+===========
+
+.. toctree::
+    :maxdepth: 1
+    :caption: Filesystem Providers
+    :glob:
+
+    *

providers/sftp/docs/filesystems/sftp.rst

@@ -0,0 +1,63 @@
+ .. 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 Filesystem
+===============
+
+Use ``ObjectStoragePath`` with SFTP/SSH servers via the `sshfs <https://github.com/fsspec/sshfs>`__ library.
+
+.. code-block:: bash
+
+    pip install apache-airflow-providers-sftp[sshfs]
+
+URL format: ``sftp://connection_id@hostname/path/to/file`` (also supports ``ssh://``).
+
+Configuration
+-------------
+
+Uses the standard SFTP connection. The following extras are supported:
+
+* ``key_file`` - path to private key file
+* ``private_key`` - private key content (PEM format)
+* ``private_key_passphrase`` - passphrase for encrypted keys
+* ``no_host_key_check`` - set to ``true`` to skip host key verification
+
+See :doc:`/connections/sftp` for details.
+
+Example
+-------
+
+.. code-block:: python
+
+    from airflow.sdk import ObjectStoragePath
+
+    path = ObjectStoragePath("sftp://my_conn@myserver/data/file.csv")
+
+    # read
+    with path.open() as f:
+        data = f.read()
+
+    # write
+    with path.open("w") as f:
+        f.write("content")
+
+    # list
+    for p in path.parent.iterdir():
+        print(p.name)
+
+    # copy
+    path.copy(ObjectStoragePath("file:///tmp/local.csv"))

providers/sftp/docs/index.rst

@@ -29,6 +29,13 @@
     Changelog <changelog>
     Security <security>
 
+.. toctree::
+    :hidden:
+    :maxdepth: 1
+    :caption: Guides
+
+    Filesystems <filesystems/index>
+
 .. toctree::
     :hidden:
     :maxdepth: 1

providers/sftp/provider.yaml

@@ -123,3 +123,6 @@ triggers:
   - integration-name: SSH File Transfer Protocol (SFTP)
     python-modules:
       - airflow.providers.sftp.triggers.sftp
+
+filesystems:
+  - airflow.providers.sftp.fs.sftp

providers/sftp/pyproject.toml

@@ -72,6 +72,9 @@ dependencies = [
 "openlineage" = [
     "apache-airflow-providers-openlineage"
 ]
+"sshfs" = [
+    "sshfs>=2023.1.0",
+]
 
 [dependency-groups]
 dev = [

providers/sftp/src/airflow/providers/sftp/fs/__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.

providers/sftp/src/airflow/providers/sftp/fs/sftp.py

@@ -0,0 +1,65 @@
+# 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 TYPE_CHECKING, Any
+
+from airflow.sdk.bases.hook import BaseHook
+
+if TYPE_CHECKING:
+    from fsspec import AbstractFileSystem
+
+schemes = ["sftp", "ssh"]
+
+
+def get_fs(conn_id: str | None, storage_options: dict[str, Any] | None = None) -> AbstractFileSystem:
+    try:
+        from sshfs import SSHFileSystem
+    except ImportError:
+        raise ImportError(
+            "Airflow FS SFTP/SSH protocol requires the sshfs library. "
+            "Install with: pip install apache-airflow-providers-sftp[sshfs]"
+        )
+
+    if conn_id is None:
+        return SSHFileSystem(**(storage_options or {}))
+
+    conn = BaseHook.get_connection(conn_id)
+    extras = conn.extra_dejson
+
+    options: dict[str, Any] = {
+        "host": conn.host,
+        "port": conn.port or 22,
+        "username": conn.login,
+    }
+
+    if conn.password:
+        options["password"] = conn.password
+
+    if key_file := extras.get("key_file"):
+        options["client_keys"] = [key_file]
+
+    if private_key := extras.get("private_key"):
+        options["client_keys"] = [private_key]
+        if passphrase := extras.get("private_key_passphrase"):
+            options["passphrase"] = passphrase
+
+    if str(extras.get("no_host_key_check", "")).lower() == "true":
+        options["known_hosts"] = None
+
+    options.update(storage_options or {})
+    return SSHFileSystem(**options)

providers/sftp/src/airflow/providers/sftp/get_provider_info.py

@@ -71,4 +71,5 @@ def get_provider_info():
                 "python-modules": ["airflow.providers.sftp.triggers.sftp"],
             }
         ],
+        "filesystems": ["airflow.providers.sftp.fs.sftp"],
     }

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