Source code for globus_sdk.services.gcs.client

import uuid
from typing import Any, Callable, Dict, Iterable, Optional, TypeVar, Union

from globus_sdk import client, response, scopes, utils
from globus_sdk.authorizers import GlobusAuthorizer
from globus_sdk.types import UUIDLike

from .data import CollectionDocument
from .errors import GCSAPIError
from .response import IterableGCSResponse, UnpackingGCSResponse

C = TypeVar("C", bound=Callable)


def _gcsdoc(message: str, link: str) -> Callable[[C], C]:
    # do not use functools.partial because it doesn't preserve type information
    # see: https://github.com/python/mypy/issues/1484
    def partial(func: C) -> C:
        return utils.doc_api_method(
            message,
            link,
            external_base_url="https://docs.globus.org/globus-connect-server/v5/api",
        )(func)

    return partial


[docs]class GCSClient(client.BaseClient): """ A GCSClient provides communication with the GCS Manager API of a Globus Connect Server instance. For full reference, see the `documentation for the GCS Manager API <https://docs.globus.org/globus-connect-server/v5/api/>`_. Unlike other client types, this must be provided with an address for the GCS Manager. All other arguments are the same as those for `~globus_sdk.BaseClient`. :param gcs_address: The FQDN (DNS name) or HTTPS URL for the GCS Manager API. :type gcs_address: str .. automethodlist:: globus_sdk.GCSClient """ service_name = "globus_connect_server" error_class = GCSAPIError def __init__( self, gcs_address: str, *, environment: Optional[str] = None, authorizer: Optional[GlobusAuthorizer] = None, app_name: Optional[str] = None, transport_params: Optional[Dict] = None, ): # check if the provided address was a DNS name or an HTTPS URL # if it was a URL, do not modify, but if it's a DNS name format it accordingly # as a heuristic for this: just check if string starts with "https://" (this is # sufficient to distinguish between the two for valid inputs) if not gcs_address.startswith("https://"): gcs_address = f"https://{gcs_address}/api/" super().__init__( base_url=gcs_address, environment=environment, authorizer=authorizer, app_name=app_name, transport_params=transport_params, )
[docs] @staticmethod def get_gcs_endpoint_scopes( endpoint_id: Union[uuid.UUID, str] ) -> scopes.GCSEndpointScopeBuilder: """Given a GCS Endpoint ID, this helper constructs an object containing the scopes for that Endpoint. :param endpoint_id: The ID of the Endpoint :type endpoint_id: UUID or str See documentation for :class:`globus_sdk.scopes.GCSEndpointScopeBuilder` for more information. """ return scopes.GCSEndpointScopeBuilder(str(endpoint_id))
[docs] @staticmethod def get_gcs_collection_scopes( collection_id: Union[uuid.UUID, str] ) -> scopes.GCSCollectionScopeBuilder: """Given a GCS Collection ID, this helper constructs an object containing the scopes for that Collection. :param collection_id: The ID of the Collection :type collection_id: UUID or str See documentation for :class:`globus_sdk.scopes.GCSCollectionScopeBuilder` for more information. """ return scopes.GCSCollectionScopeBuilder(str(collection_id))
[docs] @_gcsdoc("List Collections", "openapi_Collections/#ListCollections") def get_collection_list( self, *, include: Union[str, Iterable[str], None] = None, query_params: Optional[Dict[str, Any]] = None, ) -> IterableGCSResponse: """ ``GET /collections`` :param include: Names of additional documents to include in the response :type include: str or iterable of str, optional :param query_params: Additional passthrough query parameters :type query_params: dict, optional List the Collections on an Endpoint """ if query_params is None: query_params = {} if include is not None: if isinstance(include, str): include = [include] query_params["include"] = ",".join(include) return IterableGCSResponse(self.get("collections", query_params=query_params))
[docs] @_gcsdoc("Get Collection", "openapi_Collections/#getCollection") def get_collection( self, collection_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None, ) -> UnpackingGCSResponse: """ ``GET /collections/{collection_id}`` :param collection_id: The ID of the collection to lookup :type collection_id: str or UUID :param query_params: Additional passthrough query parameters :type query_params: dict, optional Lookup a Collection on an Endpoint """ return UnpackingGCSResponse( self.get(f"/collections/{collection_id}", query_params=query_params), "collection", )
[docs] @_gcsdoc("Create Collection", "openapi_Collections/#createCollection") def create_collection( self, collection_data: Union[Dict[str, Any], CollectionDocument], ) -> UnpackingGCSResponse: """ ``POST /collections`` Create a collection. This is used to create either a mapped or a guest collection. When created, a ``collection:administrator`` role for that collection will be created using the caller’s identity. In order to create a guest collection, the caller must have an identity that matches the Storage Gateway policies. In order to create a mapped collection, the caller must have an ``endpoint:administrator`` or ``endpoint:owner`` role. :param collection_data: The collection document for the new collection :type collection_data: dict or CollectionDocument """ return UnpackingGCSResponse( self.post("/collections", data=collection_data), "collection" )
[docs] @_gcsdoc("Update Collection", "openapi_Collections/#patchCollection") def update_collection( self, collection_id: UUIDLike, collection_data: Union[Dict[str, Any], CollectionDocument], *, query_params: Optional[Dict[str, Any]] = None, ) -> UnpackingGCSResponse: """ ``PATCH /collections/{collection_id}`` :param collection_id: The ID of the collection to update :type collection_id: str or UUID :param collection_data: The collection document for the modified collection :type collection_data: dict or CollectionDocument :param query_params: Additional passthrough query parameters :type query_params: dict, optional """ return UnpackingGCSResponse( self.patch( f"/collections/{collection_id}", data=collection_data, query_params=query_params, ), "collection", )
[docs] @_gcsdoc("Delete Collection", "openapi_Collections/#deleteCollection") def delete_collection( self, collection_id: UUIDLike, *, query_params: Optional[Dict[str, Any]] = None, ) -> response.GlobusHTTPResponse: """ ``DELETE /collections/{collection_id}`` :param collection_id: The ID of the collection to delete :type collection_id: str or UUID :param query_params: Additional passthrough query parameters :type query_params: dict, optional """ return self.delete(f"/collections/{collection_id}", query_params=query_params)