import uuid
from typing import Any, Callable, Dict, Iterable, Optional, TypeVar, Union
from globus_sdk import client, paging, response, scopes, utils
from globus_sdk.authorizers import GlobusAuthorizer
from globus_sdk.types import UUIDLike
from .data import CollectionDocument, GCSRoleDocument
from .errors import GCSAPIError
from .response import IterableGCSResponse, UnpackingGCSResponse
C = TypeVar("C", bound=Callable[..., Any])
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[str, Any]] = 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,
*,
mapped_collection_id: Optional[UUIDLike] = None,
filter: Union[ # pylint: disable=redefined-builtin
str, Iterable[str], None
] = None,
include: Union[str, Iterable[str], None] = None,
query_params: Optional[Dict[str, Any]] = None,
) -> IterableGCSResponse:
"""
``GET /collections``
:param mapped_collection_id: Filter collections which were created using this
mapped collection ID.
:type mapped_collection_id: str or UUID
:param filter: Filter the returned set to any combination of the following:
``mapped_collections``, ``guest_collections``, ``managed_by_me``,
``created_by_me``.
:type filter: str or iterable of str, optional
: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)
if mapped_collection_id is not None:
query_params["mapped_collection_id"] = mapped_collection_id
if filter is not None:
if isinstance(filter, str):
filter = [filter]
query_params["filter"] = ",".join(filter)
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)
[docs] @_gcsdoc("List Roles", "openapi_Roles/#listRoles")
@paging.has_paginator(
paging.MarkerPaginator,
items_key="data",
)
def get_role_list(
self,
collection_id: Optional[UUIDLike] = None,
include: Optional[str] = None,
query_params: Optional[Dict[str, Any]] = None,
) -> IterableGCSResponse:
"""
``GET /roles``
:param collection_id: UUID of a Collection. If given then only roles
related to that Collection are returned, otherwise only Endpoint
roles are returned.
:type collection_id: str or UUID, optional
:param include: Pass "all_roles" to request all roles all roles
relevant to the resource instead of only those the caller has on
the resource
:type include: str, optional
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
if query_params is None:
query_params = {}
if include is not None:
query_params["include"] = include
if collection_id is not None:
query_params["collection_id"] = collection_id
path = "/roles"
return IterableGCSResponse(self.get(path, query_params=query_params))
[docs] @_gcsdoc("Create Role", "openapi_Roles/#postRole")
def create_role(
self,
data: Union[Dict[str, Any], GCSRoleDocument],
query_params: Optional[Dict[str, Any]] = None,
) -> UnpackingGCSResponse:
"""
POST /roles
:param data: Data in the format of a Role document, it is recommended
to use the `RoleDocumment` class to construct this data.
:type data: dict
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = "/roles"
return UnpackingGCSResponse(
self.post(path, data=data, query_params=query_params),
"role",
)
[docs] @_gcsdoc("Get a Role", "openapi_Roles/#getRole")
def get_role(
self,
role_id: UUIDLike,
query_params: Optional[Dict[str, Any]] = None,
) -> UnpackingGCSResponse:
"""
GET /roles/{role_id}
:param role_id: UUID for the Role to be gotten
:type role_id: str or UUID
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = f"/roles/{role_id}"
return UnpackingGCSResponse(self.get(path, query_params=query_params), "role")
[docs] @_gcsdoc("Delete a Role", "openapi_Roles/#deleteRole")
def delete_role(
self,
role_id: UUIDLike,
query_params: Optional[Dict[str, Any]] = None,
) -> response.GlobusHTTPResponse:
"""
DELETE /roles/{role_id}
:param role_id: UUID for the Role to be deleted
:type role_id: str or UUID
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = f"/roles/{role_id}"
return self.delete(path, query_params=query_params)