Add support for Globus Tunnels

changelog.d/20251205_104316_john_tunnels.rst

@@ -0,0 +1,12 @@
+Added
+-----
+
+- Added support to the ``TransferClient`` for the Streams API (:pr:`NUMBER`)
+
+  - ``CreateTunnelData`` is a payload builder for tunnel creation documents
+  - ``TransferClient.create_tunnel()`` supports tunnel creation
+  - ``TransferClient.update_tunnel()`` supports updates to a tunnel
+  - ``TransferClient.get_tunnel()`` fetches a tunnel by ID
+  - ``TransferClient.delete_tunnel()`` deletes a tunnel
+  - ``TransferClient.list_tunnels()`` fetches all of the current user's tunnels
+  - ``TransferClient.get_stream_access_point()`` fetches a Stream Access Point by ID

src/globus_sdk/__init__.pyi

@@ -116,6 +116,7 @@ from .services.timers import (
     TransferTimer,
 )
 from .services.transfer import (
+    CreateTunnelData,
     DeleteData,
     IterableTransferResponse,
     TransferAPIError,

src/globus_sdk/client.py

@@ -540,7 +540,7 @@ def request(
 
         # if a client is asked to make a request against a full URL, not just the path
         # component, then do not resolve the path, simply pass it through as the URL
-        if path.startswith("https://") or path.startswith("http://"):
+        if path.startswith(("https://", "http://")):
             url = path
         else:
             url = slash_join(self.base_url, urllib.parse.quote(path))

src/globus_sdk/services/transfer/__init__.py

@@ -1,5 +1,5 @@
 from .client import TransferClient
-from .data import DeleteData, TransferData
+from .data import CreateTunnelData, DeleteData, TransferData
 from .errors import TransferAPIError
 from .response import IterableTransferResponse
 
@@ -9,4 +9,5 @@
     "DeleteData",
     "TransferAPIError",
     "IterableTransferResponse",
+    "CreateTunnelData",
 )

src/globus_sdk/services/transfer/client.py

@@ -13,7 +13,7 @@
 from globus_sdk.scopes import GCSCollectionScopes, Scope, TransferScopes
 from globus_sdk.transport import RetryConfig
 
-from .data import DeleteData, TransferData
+from .data import CreateTunnelData, DeleteData, TransferData
 from .errors import TransferAPIError
 from .response import IterableTransferResponse
 from .transport import TRANSFER_DEFAULT_RETRY_CHECKS
@@ -2699,3 +2699,195 @@ def endpoint_manager_delete_pause_rule(
             f"/v0.10/endpoint_manager/pause_rule/{pause_rule_id}",
             query_params=query_params,
         )
+
+    # Tunnel methods
+
+    def create_tunnel(
+        self,
+        data: dict[str, t.Any] | CreateTunnelData,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param data: Parameters for the tunnel creation
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    result = tc.create_tunnel(data)
+                    print(result["data"]["id"])
+
+            .. tab-item:: API Info
+
+                ``POST /v2/tunnels``
+        """
+        log.debug("TransferClient.create_tunnel(...)")
+        try:
+            data_element = data["data"]
+        except KeyError as e:
+            raise exc.GlobusSDKUsageError(
+                "create_tunnel() body was malformed (missing the 'data' key). "
+                "Use CreateTunnelData to easily create correct documents."
+            ) from e
+
+        try:
+            attributes = data_element["attributes"]
+        except KeyError:
+            data_element["attributes"] = {}
+            attributes = data_element["attributes"]
+        if attributes.get("submission_id", MISSING) is MISSING:
+            log.debug("create_tunnel auto-fetching submission_id")
+            attributes["submission_id"] = self.get_submission_id()["value"]
+
+        r = self.post("/v2/tunnels", data=data)
+        return r
+
+    def update_tunnel(
+        self,
+        tunnel_id: str,
+        update_doc: dict[str, t.Any],
+    ) -> response.GlobusHTTPResponse:
+        r"""
+        :param tunnel_id: The ID of the Tunnel.
+        :param update_doc: The document that will be sent to the patch API.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    "data" = {
+                        "type": "Tunnel",
+                        "attributes": {
+                            "state": "STOPPING",
+                        },
+                    }
+                    result = tc.update_tunnel(tunnel_id, data)
+                    print(result["data"])
+
+            .. tab-item:: API Info
+
+                ``PATCH /v2/tunnels/<tunnel_id>``
+        """
+        r = self.patch(f"/v2/tunnels/{tunnel_id}", data=update_doc)
+        return r
+
+    def get_tunnel(
+        self,
+        tunnel_id: str,
+        *,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param tunnel_id: The ID of the Tunnel which we are fetching details about.
+        :param query_params: Any additional parameters will be passed through
+            as query params.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    result = tc.show_tunnel(tunnel_id)
+                    print(result["data"])
+
+            .. tab-item:: API Info
+
+                ``GET /v2/tunnels/<tunnel_id>``
+        """
+        log.debug("TransferClient.get_tunnel(...)")
+        r = self.get(f"/v2/tunnels/{tunnel_id}", query_params=query_params)
+        return r
+
+    def delete_tunnel(
+        self,
+        tunnel_id: str,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param tunnel_id: The ID of the Tunnel to be deleted.
+
+        This will clean up all data associated with a Tunnel.
+        Note that Tunnels must be stopped before they can be deleted.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    tc.delete_tunnel(tunnel_id)
+
+            .. tab-item:: API Info
+
+                ``DELETE /v2/tunnels/<tunnel_id>``
+        """
+        log.debug("TransferClient.delete_tunnel(...)")
+        r = self.delete(f"/v2/tunnels/{tunnel_id}")
+        return r
+
+    def list_tunnels(
+        self,
+        *,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> IterableTransferResponse:
+        """
+        :param query_params: Any additional parameters will be passed through
+            as query params.
+
+        This will list all the Tunnels created by the authorized user.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    tc.list_tunnels(tunnel_id)
+
+            .. tab-item:: API Info
+
+                ``GET /v2/tunnels/``
+        """
+        log.debug("TransferClient.list_tunnels(...)")
+        r = self.get("/v2/tunnels", query_params=query_params)
+        return IterableTransferResponse(r)
+
+    def get_stream_access_point(
+        self,
+        stream_ap_id: str,
+        *,
+        query_params: dict[str, t.Any] | None = None,
+    ) -> response.GlobusHTTPResponse:
+        """
+        :param stream_ap_id: The ID of the steaming access point to lookup.
+        :param query_params: Any additional parameters will be passed through
+            as query params.
+
+        This will list all the Tunnels created by the authorized user.
+
+        .. tab-set::
+
+            .. tab-item:: Example Usage
+
+                .. code-block:: python
+
+                    tc = globus_sdk.TunnelClient(...)
+                    tc.get_stream_ap(stream_ap_id)
+
+            .. tab-item:: API Info
+
+                ``GET /v2/stream_access_points/<stream_ap_id>``
+        """
+        log.debug("TransferClient.get_stream_ap(...)")
+        r = self.get(
+            f"/v2/stream_access_points/{stream_ap_id}", query_params=query_params
+        )
+        return r

src/globus_sdk/services/transfer/data/__init__.py

@@ -6,5 +6,6 @@
 
 from .delete_data import DeleteData
 from .transfer_data import TransferData
+from .tunnel_data import CreateTunnelData
 
-__all__ = ("TransferData", "DeleteData")
+__all__ = ("TransferData", "DeleteData", "CreateTunnelData")

src/globus_sdk/services/transfer/data/tunnel_data.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import logging
+import typing as t
+import uuid
+
+from globus_sdk._missing import MISSING, MissingType
+from globus_sdk._payload import GlobusPayload
+
+log = logging.getLogger(__name__)
+
+
+class CreateTunnelData(GlobusPayload):
+    def __init__(
+        self,
+        initiator_stream_access_point: uuid.UUID | str,
+        listener_stream_access_point: uuid.UUID | str,
+        *,
+        label: str | MissingType = MISSING,
+        submission_id: uuid.UUID | str | MissingType = MISSING,
+        lifetime_mins: int | MissingType = MISSING,
+        restartable: bool | MissingType = MISSING,
+        additional_fields: dict[str, t.Any] | None = None,
+    ) -> None:
+        super().__init__()
+        log.debug("Creating a new TunnelData object")
+
+        relationships = {
+            "listener": {
+                "data": {
+                    "type": "StreamAccessPoint",
+                    "id": listener_stream_access_point,
+                }
+            },
+            "initiator": {
+                "data": {
+                    "type": "StreamAccessPoint",
+                    "id": initiator_stream_access_point,
+                }
+            },
+        }
+        attributes = {
+            "label": label,
+            "submission_id": submission_id,
+            "restartable": restartable,
+            "lifetime_mins": lifetime_mins,
+        }
+        if additional_fields is not None:
+            attributes.update(additional_fields)
+
+        self["data"] = {
+            "type": "Tunnel",
+            "relationships": relationships,
+            "attributes": attributes,
+        }

src/globus_sdk/testing/data/transfer/create_tunnel.py

@@ -0,0 +1,59 @@
+import uuid
+
+from globus_sdk.testing.models import RegisteredResponse, ResponseSet
+
+TUNNEL_ID = str(uuid.uuid4())
+
+_initiator_ap = str(uuid.uuid4())
+_listener_ap = str(uuid.uuid4())
+
+_default_display_name = "Test Tunnel"
+
+
+RESPONSES = ResponseSet(
+    default=RegisteredResponse(
+        service="transfer",
+        method="POST",
+        path="/v2/tunnels",
+        json={
+            "data": {
+                "attributes": {
+                    "created_time": "2025-12-12T21:49:22.183977",
+                    "initiator_ip_address": None,
+                    "initiator_port": None,
+                    "label": _default_display_name,
+                    "lifetime_mins": 10,
+                    "listener_ip_address": None,
+                    "listener_port": None,
+                    "restartable": False,
+                    "state": "AWAITING_LISTENER",
+                    "status": "The tunnel is waiting for listening.",
+                    "submission_id": "6ab42cda-d7a4-11f0-ad34-0affc202d2e9",
+                },
+                "id": "34d97133-f17e-4f90-ad42-56ff5f3c2550",
+                "relationships": {
+                    "initiator": {
+                        "data": {"id": _initiator_ap, "type": "StreamAccessPoint"}
+                    },
+                    "listener": {
+                        "data": {"id": _listener_ap, "type": "StreamAccessPoint"}
+                    },
+                    "owner": {
+                        "data": {
+                            "id": "4d443580-012d-4954-816f-e0592bd356e1",
+                            "type": "Identity",
+                        }
+                    },
+                },
+                "type": "Tunnel",
+            },
+            "meta": {"request_id": "e6KkKkNmw"},
+        },
+        metadata={
+            "tunnel_id": TUNNEL_ID,
+            "display_name": _default_display_name,
+            "initiator_ap": _initiator_ap,
+            "listener_ap": _listener_ap,
+        },
+    ),
+)

src/globus_sdk/testing/data/transfer/delete_tunnel.py

@@ -0,0 +1,18 @@
+import uuid
+
+from globus_sdk.testing.models import RegisteredResponse, ResponseSet
+
+TUNNEL_ID = str(uuid.uuid4())
+
+
+RESPONSES = ResponseSet(
+    default=RegisteredResponse(
+        service="transfer",
+        method="DELETE",
+        path=f"/v2/tunnels/{TUNNEL_ID}",
+        json={"data": None, "meta": {"request_id": "ofayi2B4R"}},
+        metadata={
+            "tunnel_id": TUNNEL_ID,
+        },
+    ),
+)