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)