<filename>src/globus_cli/services/transfer/client.py
import logging
import textwrap
import uuid
from typing import Any, Dict, Tuple, Union

import click
from globus_sdk import GlobusHTTPResponse, TransferClient

from .data import display_name_or_cname
from .recursive_ls import RecursiveLsResponse

log = logging.getLogger(__name__)


class CustomTransferClient(TransferClient):
    # TODO: Remove this function when endpoints natively support recursive ls
    def recursive_operation_ls(
        self,
        endpoint_id: Union[str, uuid.UUID],
        params: Dict[str, Any],
        depth: int = 3,
    ) -> RecursiveLsResponse:
        """
        Makes recursive calls to ``GET /operation/endpoint/<endpoint_id>/ls``
        Does not preserve access to top level operation_ls fields, but
        adds a "path" field for every item that represents the full
        path to that item.

        :rtype: iterable of :class:`GlobusResponse <globus_sdk.response.GlobusResponse>`

        :param endpoint_id: The endpoint being recursively ls'ed. If no "path" is given
            in params, the start path is determined by this endpoint.
        :param params: Parameters that will be passed through as query params.
        :param depth: The maximum file depth the recursive ls will go to.
        """
        endpoint_id = str(endpoint_id)
        log.info(
            "TransferClient.recursive_operation_ls(%s, %s, %s)",
            endpoint_id,
            depth,
            params,
        )
        return RecursiveLsResponse(self, endpoint_id, params, max_depth=depth)

    def get_endpoint_w_server_list(
        self, endpoint_id
    ) -> Tuple[GlobusHTTPResponse, Union[str, GlobusHTTPResponse]]:
        """
        A helper for handling endpoint server list lookups correctly accounting
        for various endpoint types.

        - Raises click.UsageError when used on Shares
        - Returns (<get_endpoint_response>, "S3") for S3 endpoints
        - Returns (<get_endpoint_response>, <server_list_response>) for all other
          Endpoints
        """
        endpoint = self.get_endpoint(endpoint_id)

        if endpoint["host_endpoint_id"]:  # not GCS -- this is a share endpoint
            raise click.UsageError(
                textwrap.dedent(
                    """\
                {id} ({0}) is a share and does not have servers.

                To see details of the share, use
                    globus endpoint show {id}

                To list the servers on the share's host endpoint, use
                    globus endpoint server list {host_endpoint_id}
            """
                ).format(display_name_or_cname(endpoint), **endpoint.data)
            )

        if endpoint["s3_url"]:  # not GCS -- legacy S3 endpoint type
            return (endpoint, "S3")

        else:
            return (endpoint, self.endpoint_server_list(endpoint_id))

<filename>src/globus_cli/services/transfer/client.py
import logging
import textwrap
import uuid
from typing import Any, Dict, Tuple, Union

import click
from globus_sdk import GlobusHTTPResponse, TransferClient

from .data import display_name_or_cname
from .recursive_ls import RecursiveLsResponse

log = logging.getLogger(__name__)


class CustomTransferClient(TransferClient):
    # TODO: Remove this function when endpoints natively support recursive ls
    def recursive_operation_ls(
        self,
        endpoint_id: Union[str, uuid.UUID],
        params: Dict[str, Any],
        depth: int = 3,
    ) -> RecursiveLsResponse:
        """
        Makes recursive calls to ``GET /operation/endpoint/<endpoint_id>/ls``
        Does not preserve access to top level operation_ls fields, but
        adds a "path" field for every item that represents the full
        path to that item.

        :rtype: iterable of :class:`GlobusResponse <globus_sdk.response.GlobusResponse>`

        :param endpoint_id: The endpoint being recursively ls'ed. If no "path" is given
            in params, the start path is determined by this endpoint.
        :param params: Parameters that will be passed through as query params.
        :param depth: The maximum file depth the recursive ls will go to.
        """
        endpoint_id = str(endpoint_id)
        log.info(
            "TransferClient.recursive_operation_ls(%s, %s, %s)",
            endpoint_id,
            depth,
            params,
        )
        return RecursiveLsResponse(self, endpoint_id, params, max_depth=depth)

    def get_endpoint_w_server_list(
        self, endpoint_id
    ) -> Tuple[GlobusHTTPResponse, Union[str, GlobusHTTPResponse]]:
        """
        A helper for handling endpoint server list lookups correctly accounting
        for various endpoint types.

        - Raises click.UsageError when used on Shares
        - Returns (<get_endpoint_response>, "S3") for S3 endpoints
        - Returns (<get_endpoint_response>, <server_list_response>) for all other
          Endpoints
        """
        endpoint = self.get_endpoint(endpoint_id)

        if endpoint["host_endpoint_id"]:  # not GCS -- this is a share endpoint
            raise click.UsageError(
                textwrap.dedent(
                    """\
                {id} ({0}) is a share and does not have servers.

                To see details of the share, use
                    globus endpoint show {id}

                To list the servers on the share's host endpoint, use
                    globus endpoint server list {host_endpoint_id}
            """
                ).format(display_name_or_cname(endpoint), **endpoint.data)
            )

        if endpoint["s3_url"]:  # not GCS -- legacy S3 endpoint type
            return (endpoint, "S3")

        else:
            return (endpoint, self.endpoint_server_list(endpoint_id))

# TODO: Remove this function when endpoints natively support recursive ls Makes recursive calls to ``GET /operation/endpoint/<endpoint_id>/ls``         Does not preserve access to top level operation_ls fields, but         adds a "path" field for every item that represents the full         path to that item.          :rtype: iterable of :class:`GlobusResponse <globus_sdk.response.GlobusResponse>`          :param endpoint_id: The endpoint being recursively ls'ed. If no "path" is given             in params, the start path is determined by this endpoint.         :param params: Parameters that will be passed through as query params.         :param depth: The maximum file depth the recursive ls will go to. A helper for handling endpoint server list lookups correctly accounting         for various endpoint types.          - Raises click.UsageError when used on Shares         - Returns (<get_endpoint_response>, "S3") for S3 endpoints         - Returns (<get_endpoint_response>, <server_list_response>) for all other           Endpoints # not GCS -- this is a share endpoint \                 {id} ({0}) is a share and does not have servers.                  To see details of the share, use                     globus endpoint show {id}                  To list the servers on the share's host endpoint, use                     globus endpoint server list {host_endpoint_id} # not GCS -- legacy S3 endpoint type