import typing as t
import uuid
from globus_sdk import client, paging, response, scopes, utils
from globus_sdk._types import UUIDLike
from globus_sdk.authorizers import GlobusAuthorizer
from .data import (
CollectionDocument,
GCSRoleDocument,
StorageGatewayDocument,
UserCredentialDocument,
)
from .errors import GCSAPIError
from .response import IterableGCSResponse, UnpackingGCSResponse
C = t.TypeVar("C", bound=t.Callable[..., t.Any])
def _gcsdoc(message: str, link: str) -> t.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: t.Optional[str] = None,
authorizer: t.Optional[GlobusAuthorizer] = None,
app_name: t.Optional[str] = None,
transport_params: t.Optional[t.Dict[str, t.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: t.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: t.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] @staticmethod
def connector_id_to_name(connector_id: UUIDLike) -> t.Optional[str]:
"""
Helper that converts a given connector_id into a human readable
connector name string. Will return None if the id is not recognized.
Note that it is possible for valid connector_ids to be unrecognized
due to differing SDK and GCS versions.
"""
connector_dict = {
"7c100eae-40fe-11e9-95a3-9cb6d0d9fd63": "Box",
"1b6374b0-f6a4-4cf7-a26f-f262d9c6ca72": "Ceph",
"56366b96-ac98-11e9-abac-9cb6d0d9fd63": "Google Cloud Storage",
"976cf0cf-78c3-4aab-82d2-7c16adbcc281": "Google Drive",
"145812c8-decc-41f1-83cf-bb2a85a2a70b": "POSIX",
"7643e831-5f6c-4b47-a07f-8ee90f401d23": "S3",
"7e3f3f5e-350c-4717-891a-2f451c24b0d4": "SpectraLogic BlackPearl",
}
return connector_dict.get(str(connector_id))
#
# collection methods
#
[docs] @_gcsdoc("List Collections", "openapi_Collections/#ListCollections")
def get_collection_list(
self,
*,
mapped_collection_id: t.Optional[UUIDLike] = None,
filter: t.Union[ # pylint: disable=redefined-builtin
str, t.Iterable[str], None
] = None,
include: t.Union[str, t.Iterable[str], None] = None,
query_params: t.Optional[t.Dict[str, t.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:
query_params["include"] = ",".join(utils.safe_strseq_iter(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: t.Optional[t.Dict[str, t.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: t.Union[t.Dict[str, t.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: t.Union[t.Dict[str, t.Any], CollectionDocument],
*,
query_params: t.Optional[t.Dict[str, t.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: t.Optional[t.Dict[str, t.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)
#
# storage gateway methods
#
[docs] @_gcsdoc("List Storage Gateways", "openapi_Storage_Gateways/#getStorageGateways")
@paging.has_paginator(
paging.MarkerPaginator,
items_key="data",
)
def get_storage_gateway_list(
self,
*,
include: t.Union[None, str, t.Iterable[str]] = None,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> IterableGCSResponse:
"""
``GET /storage_gateways``
:param include: Optional document types to include in the response. If
'private_policies' is included, then include private storage gateway
policies in the attached storage_gateways document. This requires an
``administrator`` role on the Endpoint.
:type include: str or iterable of 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"] = ",".join(utils.safe_strseq_iter(include))
return IterableGCSResponse(
self.get("/storage_gateways", query_params=query_params)
)
[docs] @_gcsdoc("Create a Storage Gateway", "openapi_Storage_Gateways/#postStorageGateway")
def create_storage_gateway(
self,
data: t.Union[t.Dict[str, t.Any], StorageGatewayDocument],
*,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
``POST /storage_gateways``
:param data: Data in the format of a Storage Gateway document, it is recommended
to use the ``StorageGatewayDocumment`` class to construct this data.
:type data: dict or StorageGatewayDocument
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
return UnpackingGCSResponse(
self.post("/storage_gateways", data=data, query_params=query_params),
"storage_gateway",
)
[docs] @_gcsdoc("Get a Storage Gateway", "openapi_Storage_Gateways/#getStorageGateway")
def get_storage_gateway(
self,
storage_gateway_id: UUIDLike,
*,
include: t.Union[None, str, t.Iterable[str]] = None,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
``GET /storage_gateways/<storage_gateway_id>``
:param storage_gateway_id: UUID for the Storage Gateway to be gotten
:type storage_gateway_id: str or UUID
:param include: Optional document types to include in the response. If
'private_policies' is included, then include private storage gateway
policies in the attached storage_gateways document. This requires an
``administrator`` role on the Endpoint.
:type include: str or iterable of 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"] = ",".join(utils.safe_strseq_iter(include))
return UnpackingGCSResponse(
self.get(
f"/storage_gateways/{storage_gateway_id}",
query_params=query_params,
),
"storage_gateway",
)
[docs] @_gcsdoc(
"Update a Storage Gateway", "openapi_Storage_Gateways/#patchStorageGateway"
)
def update_storage_gateway(
self,
storage_gateway_id: UUIDLike,
data: t.Union[t.Dict[str, t.Any], StorageGatewayDocument],
*,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> response.GlobusHTTPResponse:
"""
``PATCH /storage_gateways/<storage_gateway_id>``
:param storage_gateway_id: UUID for the Storage Gateway to be updated
:type storage_gateway_id: str or UUID
:param data: Data in the format of a Storage Gateway document, it is recommended
to use the ``StorageGatewayDocumment`` class to construct this data.
:type data: dict or StorageGatewayDocument
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
return self.patch(
f"/storage_gateways/{storage_gateway_id}",
data=data,
query_params=query_params,
)
[docs] @_gcsdoc(
"Delete a Storage Gateway", "openapi_Storage_Gateways/#deleteStorageGateway"
)
def delete_storage_gateway(
self,
storage_gateway_id: t.Union[str, uuid.UUID],
*,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> response.GlobusHTTPResponse:
"""
``DELETE /storage_gateways/<storage_gateway_id>``
:param storage_gateway_id: UUID for the Storage Gateway to be deleted
:type storage_gateway_id: str or UUID
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
return self.delete(
f"/storage_gateways/{storage_gateway_id}", query_params=query_params
)
#
# role methods
#
[docs] @_gcsdoc("List Roles", "openapi_Roles/#listRoles")
@paging.has_paginator(
paging.MarkerPaginator,
items_key="data",
)
def get_role_list(
self,
collection_id: t.Optional[UUIDLike] = None,
include: t.Optional[str] = None,
query_params: t.Optional[t.Dict[str, t.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: t.Union[t.Dict[str, t.Any], GCSRoleDocument],
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
POST /roles
:param data: Data in the format of a Role document, it is recommended
to use the `GCSRoleDocumment` 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: t.Optional[t.Dict[str, t.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: t.Optional[t.Dict[str, t.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)
[docs] @_gcsdoc("Get User Credential list", "openapi_User_Credentials/#getUserCredentials")
def get_user_credential_list(
self,
storage_gateway: t.Optional[UUIDLike] = None,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> IterableGCSResponse:
"""
GET /user_credentials
:param storage_gateway: UUID of a storage gateway to limit results to
:type storage_gateway: str or UUID, optional
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
if query_params is None:
query_params = {}
if storage_gateway is not None:
query_params["storage_gateway"] = storage_gateway
path = "/user_credentials"
return IterableGCSResponse(self.get(path, query_params=query_params))
[docs] @_gcsdoc("Create a User Credential", "openapi_User_Credentials/#postUserCredential")
def create_user_credential(
self,
data: t.Union[t.Dict[str, t.Any], UserCredentialDocument],
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
POST /user_credentials
:param data: Data in the format of a UserCredential document, it is
recommended to use the `UserCredential` class to construct this
:type data: dict
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = "/user_credentials"
return UnpackingGCSResponse(
self.post(path, data=data, query_params=query_params),
"user_credential",
)
[docs] @_gcsdoc("Get a User Credential", "openapi_User_Credentials/#getUserCredential")
def get_user_credential(
self,
user_credential_id: UUIDLike,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
GET /user_credentials/{user_credential_id}
:param user_credential_id: UUID for the UserCredential to be gotten
:type user_credential_id: str or UUID
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = f"/user_credentials/{user_credential_id}"
return UnpackingGCSResponse(
self.get(path, query_params=query_params), "user_credential"
)
[docs] @_gcsdoc(
"Update a User Credential", "openapi_User_Credentials/#patchUserCredential"
)
def update_user_credential(
self,
user_credential_id: UUIDLike,
data: t.Union[t.Dict[str, t.Any], UserCredentialDocument],
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> UnpackingGCSResponse:
"""
PATCH /user_credentials/{user_credential_id}
:param user_credential_id: UUID for the UserCredential to be updated
:type user_credential_id: str or UUID
:param data: Data in the format of a UserCredential document, it is
recommended to use the `UserCredential` class to construct this
:type data: dict
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = f"/user_credentials/{user_credential_id}"
return UnpackingGCSResponse(
self.patch(path, data=data, query_params=query_params), "user_credential"
)
[docs] @_gcsdoc(
"Delete a User Credential", "openapi_User_Credentials/#deleteUserCredential"
)
def delete_user_credential(
self,
user_credential_id: UUIDLike,
query_params: t.Optional[t.Dict[str, t.Any]] = None,
) -> response.GlobusHTTPResponse:
"""
DELETE /user_credentials/{user_credential_id}
:param user_credential_id: UUID for the UserCredential to be deleted
:type user_credential_id: str or UUID
:param query_params: Additional passthrough query parameters
:type query_params: dict, optional
"""
path = f"/user_credentials/{user_credential_id}"
return self.delete(path, query_params=query_params)