Adding commands for Streaming

Author: buzztroll

pyproject.toml

@@ -93,7 +93,7 @@ attr = "globus_cli.version.__version__"
 profile = "black"
 
 [tool.pytest.ini_options]
-addopts = "-n auto --maxprocesses 4 --dist loadscope --timeout 3 --color=yes"
+#addopts = "-n auto --maxprocesses 4 --dist loadscope --timeout 3 --color=yes"
 filterwarnings = [
     "error",
 ]

src/__init__.py

@@ -27,6 +27,7 @@
         "task": ("task", "task_command"),
         "timer": ("timer", "timer_command"),
         "transfer": ("transfer", "transfer_command"),
+        "streams": ("streams", "stream_command"),
         "update": ("update", "update_command"),
         "version": ("version", "version_command"),
         "whoami": ("whoami", "whoami_command"),

src/globus_cli/commands/__init__.py

@@ -0,0 +1,43 @@
+import typing as t
+
+from globus_cli.parsing import group
+
+
+def streams_flatten_response(in_doc: dict[str, t.Any]) -> dict[str, list[t.Any]]:
+    output_dict: dict[str, list[t.Any]] = {}
+    attributes_doc = in_doc.pop("attributes")
+    for i in attributes_doc:
+        output_dict[f"attributes_{i}"] = attributes_doc.get(i)
+    for i in in_doc:
+        output_dict[i] = in_doc.get(i)
+    relationships = in_doc.get("relationships")
+    for o in relationships:
+        inner = relationships.get(o).get("data")
+        for i in inner:
+            output_dict[f"relationships_{o}_{i}"] = relationships.get(i)
+    return output_dict
+
+
+def streams_display_flatten_response(
+    in_doc: dict[str, t.Any],
+) -> dict[str, list[t.Any]]:
+    return streams_flatten_response(in_doc["data"])
+
+
+def streams_display_response_status(in_doc: dict[str, t.Any]) -> str:
+    data_doc = in_doc["data"]
+    attributes_doc = data_doc.get("attributes")
+    if attributes_doc:
+        return attributes_doc["status"]
+    return "unknown"
+
+
+@group(
+    "streams",
+    lazy_subcommands={
+        "tunnel": (".tunnel", "tunnel_command"),
+        "access-point": (".access_point", "access_point_command"),
+    },
+)
+def stream_command() -> None:
+    """Manage Globus Tunnels."""

src/globus_cli/commands/streams/__init__.py

@@ -0,0 +1,11 @@
+from globus_cli.parsing import group
+
+
+@group(
+    "access-point",
+    lazy_subcommands={
+        "show": (".show", "show_access_point_command"),
+    },
+)
+def access_point_command() -> None:
+    """Manage Globus Tunnels."""

src/globus_cli/commands/streams/access_point/__init__.py

@@ -0,0 +1,42 @@
+import uuid
+
+import click
+
+from globus_cli.commands.streams import streams_display_flatten_response
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import command
+from globus_cli.termio import Field, display
+
+STANDARD_FIELDS = [
+    Field("ID", "id"),
+    Field("Display Name", "attributes_display_name"),
+    Field("Contact Email", "attributes_contact_email"),
+    Field("Contact Info", "attributes_contact_info"),
+    Field("Department", "attributes_department"),
+    Field("Description", "attributes_description"),
+    Field("Info Link", "attributes_info_link"),
+    Field("Organization", "attributes_organization"),
+]
+
+
+@command("show", short_help="Show the attributes of a stream access point.")
+@click.argument(
+    "stream_access_point_id", metavar="STREAM_ACCESS_POINT_ID", type=click.UUID
+)
+@LoginManager.requires_login("auth", "transfer")
+def show_access_point_command(
+    login_manager: LoginManager,
+    *,
+    stream_access_point_id: uuid.UUID,
+) -> None:
+    """
+    Create a Globus Tunnel.
+    """
+    tunnel_client = login_manager.get_transfer_client()
+    res = tunnel_client.get_stream_ap(stream_access_point_id)
+    display(
+        res,
+        text_mode=display.RECORD,
+        fields=STANDARD_FIELDS,
+        response_key=streams_display_flatten_response,
+    )

src/globus_cli/commands/streams/access_point/show.py

@@ -0,0 +1,16 @@
+from globus_cli.parsing import group
+
+
+@group(
+    "tunnel",
+    lazy_subcommands={
+        "create": (".create", "create_tunnel_command"),
+        "update": (".update", "update_tunnel_command"),
+        "delete": (".delete", "delete_tunnel_command"),
+        "list": (".list", "list_tunnel_command"),
+        "show": (".show", "show_tunnel_command"),
+        "stop": (".stop", "stop_tunnel_command"),
+    },
+)
+def tunnel_command() -> None:
+    """Manage Globus Tunnels."""

src/globus_cli/commands/streams/tunnel/__init__.py

@@ -0,0 +1,74 @@
+import uuid
+
+import click
+from globus_sdk.services.transfer.data.tunnel_data import CreateTunnelData
+
+from globus_cli.commands.streams import streams_display_flatten_response
+from globus_cli.constants import ExplicitNullType
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import command
+from globus_cli.termio import Field, display
+
+STANDARD_FIELDS = [
+    Field("ID", "id"),
+    Field("Label", "attributes_label"),
+    Field("State", "attributes_state"),
+    Field("Status", "attributes_status"),
+]
+
+
+@command("create", short_help="Create a Globus Tunnel.")
+@click.argument(
+    "initiator_stream_ap_id", metavar="INITIATOR_STREAM_AP_ID", type=click.UUID
+)
+@click.argument(
+    "listener_stream_ap_id", metavar="LISTENER_STREAM_AP_ID", type=click.UUID
+)
+@click.option(
+    "--label",
+    type=str,
+    default=None,
+    help="A friendly label to associate with the Tunnel.",
+)
+@click.option(
+    "--lifetime-minutes",
+    type=int,
+    help="The amount of time in minutes that the tunnel will exist.",
+)
+@click.option(
+    "--restartable",
+    type=bool,
+    default=False,
+    help=(
+        "A boolean to indicate if the tunnel should be restart in the "
+        "event of an error."
+    ),
+)
+@LoginManager.requires_login("auth", "transfer")
+def create_tunnel_command(
+    login_manager: LoginManager,
+    *,
+    initiator_stream_ap_id: uuid.UUID,
+    listener_stream_ap_id: uuid.UUID,
+    label: str | None,
+    lifetime_minutes: int | None,
+    restartable: bool,
+) -> None:
+    """
+    Create a Globus Tunnel.
+    """
+    data = CreateTunnelData(
+        initiator_stream_ap_id,
+        listener_stream_ap_id,
+        label=label,
+        lifetime_mins=lifetime_minutes,
+        restartable=restartable,
+    )
+    tunnel_client = login_manager.get_transfer_client()
+    res = tunnel_client.create_tunnel(data)
+    display(
+        res,
+        text_mode=display.RECORD,
+        fields=STANDARD_FIELDS,
+        response_key=streams_display_flatten_response,
+    )

src/globus_cli/commands/streams/tunnel/create.py

@@ -0,0 +1,24 @@
+import uuid
+
+import click
+
+from globus_cli.commands.streams import streams_display_response_status
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import command
+from globus_cli.termio import display
+
+
+@command("delete", short_help="Delete a Globus Tunnel.")
+@click.argument("tunnel_id", metavar="TUNNEL_ID", type=click.UUID)
+@LoginManager.requires_login("auth", "transfer")
+def delete_tunnel_command(
+    login_manager: LoginManager,
+    *,
+    tunnel_id: uuid.UUID,
+) -> None:
+    """
+    Delete a Globus Tunnel document.
+    """
+    tunnel_client = login_manager.get_transfer_client()
+    res = tunnel_client.delete_tunnel(str(tunnel_id))
+    display(res, text_mode=display.SILENT, response_key=streams_display_response_status)

src/globus_cli/commands/streams/tunnel/delete.py

@@ -0,0 +1,40 @@
+import typing as t
+
+from globus_cli.commands.streams import streams_flatten_response
+from globus_cli.login_manager import LoginManager
+from globus_cli.parsing import command
+from globus_cli.termio import Field, display
+
+STANDARD_FIELDS = [
+    Field("ID", "id"),
+    Field("Label", "attributes_label"),
+    Field("State", "attributes_state"),
+    Field("Created Time", "attributes_created_time"),
+]
+
+
+def tunnel_response_to_list(in_doc: dict[str, t.Any]) -> list[t.Any]:
+    out_list = []
+    data_doc: list[dict[str, t.Any]] = in_doc["data"]
+    for ent in data_doc:
+        new_ent = streams_flatten_response(ent)
+        out_list.append(new_ent)
+    return out_list
+
+
+@command("list", short_help="List Globus Tunnels.")
+@LoginManager.requires_login("auth", "transfer")
+def list_tunnel_command(
+    login_manager: LoginManager,
+) -> None:
+    """
+    List all the Globus Tunnels that this user can see.
+    """
+    tunnel_client = login_manager.get_transfer_client()
+    res = tunnel_client.list_tunnels()
+    display(
+        res,
+        text_mode=display.TABLE,
+        fields=STANDARD_FIELDS,
+        response_key=tunnel_response_to_list,
+    )