tessl/pypi-globus-sdk

Python SDK for interacting with Globus web APIs including Transfer, Auth, and other research data management services

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Globus Connect Server Management

Name: tessl/pypi-globus-sdk
Author: tessl

Configuration and management of Globus Connect Server endpoints, collections, storage gateways, and access policies for institutional data sharing. The GCS service provides comprehensive tools for setting up and managing data access infrastructure with fine-grained control over permissions, storage backends, and user credentials.

Capabilities

GCS Client

Core client for Globus Connect Server management operations with comprehensive endpoint, collection, and storage gateway administration.

class GCSClient(BaseClient):
    """
    Client for GCS Manager API operations.
    
    Manages Globus Connect Server instances including endpoints, collections,
    storage gateways, roles, and user credentials with comprehensive
    configuration and policy support.
    """
    
    def __init__(
        self,
        gcs_address: str,
        *,
        app: GlobusApp | None = None,
        app_scopes: list[Scope] | None = None,
        environment: str | None = None,
        authorizer: GlobusAuthorizer | None = None,
        app_name: str | None = None,
        transport_params: dict[str, Any] | None = None
    ) -> None: ...
    
    @staticmethod
    def get_gcs_endpoint_scopes(
        endpoint_id: str | UUID
    ) -> GCSEndpointScopeBuilder:
        """
        Get scope builder for GCS endpoint-specific operations.
        
        Creates a scope builder for generating scopes required for
        endpoint management operations on a specific GCS endpoint.
        
        Parameters:
        - endpoint_id: UUID of the GCS endpoint
        
        Returns:
        GCSEndpointScopeBuilder for constructing endpoint-specific scopes
        """
    
    @staticmethod
    def get_gcs_collection_scopes(
        collection_id: str | UUID
    ) -> GCSCollectionScopeBuilder:
        """
        Get scope builder for GCS collection-specific operations.
        
        Creates a scope builder for generating scopes required for
        collection management and data access operations.
        
        Parameters:
        - collection_id: UUID of the GCS collection
        
        Returns:
        GCSCollectionScopeBuilder for constructing collection-specific scopes
        """

Endpoint Management

Manage Globus Connect Server endpoint configuration and metadata for institutional data sharing infrastructure.

def get_gcs_info(
    self, 
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get general information about the GCS instance.
    
    Returns basic configuration information, capabilities, and
    metadata about the Globus Connect Server installation.
    
    Parameters:
    - query_params: Additional query parameters
    
    Returns:
    UnpackingGCSResponse with GCS instance information
    """

def get_endpoint(
    self, 
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get endpoint configuration and metadata.
    
    Returns complete endpoint information including display name,
    description, organization, contact info, and other metadata.
    
    Parameters:
    - query_params: Additional query parameters
    
    Returns:
    UnpackingGCSResponse with endpoint configuration
    """

def update_endpoint(
    self,
    endpoint_data: dict[str, Any] | EndpointDocument,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Update endpoint configuration.
    
    Modifies endpoint metadata including display name, description,
    organization information, contact details, and other settings.
    
    Parameters:
    - endpoint_data: Endpoint configuration updates
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming update
    """

Collection Management

Create and manage collections for organized data access with support for mapped collections, guest collections, and comprehensive policy configuration.

def get_collection_list(
    self,
    *,
    mapped_collection_id: str | UUID | None = None,
    include: str | Iterable[str] | None = None,
    query_params: dict[str, Any] | None = None
) -> IterableGCSResponse:
    """
    List collections with filtering and expansion options.
    
    Parameters:
    - mapped_collection_id: Filter by parent mapped collection
    - include: Additional data to include (user_credential_id, etc.)
    - query_params: Additional query parameters
    
    Returns:
    IterableGCSResponse with paginated collection listings
    """

def get_collection(
    self,
    collection_id: str | UUID,
    *,
    include: str | Iterable[str] | None = None,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get detailed collection configuration.
    
    Parameters:
    - collection_id: UUID of the collection
    - include: Additional data to include in response
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with complete collection configuration
    """

def create_collection(
    self,
    collection_data: dict[str, Any] | CollectionDocument
) -> UnpackingGCSResponse:
    """
    Create a new collection.
    
    Creates either a mapped collection (direct storage access) or
    guest collection (restricted access) with specified policies
    and access controls.
    
    Parameters:
    - collection_data: Complete collection configuration
    
    Returns:
    UnpackingGCSResponse with created collection details
    """

def update_collection(
    self,
    collection_id: str | UUID,
    collection_data: dict[str, Any] | CollectionDocument,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Update collection configuration.
    
    Modifies collection settings including display name, description,
    policies, and access controls.
    
    Parameters:
    - collection_id: UUID of collection to update
    - collection_data: Updated configuration
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming update
    """

def delete_collection(
    self,
    collection_id: str | UUID,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Delete a collection.
    
    Permanently removes a collection. Data remains on storage
    but is no longer accessible through Globus services.
    
    Parameters:
    - collection_id: UUID of collection to delete
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming deletion
    """

Storage Gateway Management

Manage storage gateways that provide the connection between collections and underlying storage systems with support for multiple storage backends.

def get_storage_gateway_list(
    self,
    *,
    include: str | Iterable[str] | None = None,
    query_params: dict[str, Any] | None = None
) -> IterableGCSResponse:
    """
    List storage gateways with optional data expansion.
    
    Parameters:
    - include: Additional data to include (policies, etc.)
    - query_params: Additional parameters
    
    Returns:
    IterableGCSResponse with paginated storage gateway listings
    """

def create_storage_gateway(
    self,
    data: dict[str, Any] | StorageGatewayDocument,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Create a new storage gateway.
    
    Creates a storage gateway connecting to underlying storage
    infrastructure (POSIX filesystems, S3, etc.) with specified
    connector type and policies.
    
    Parameters:
    - data: Storage gateway configuration including connector and policies
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with created gateway details
    """

def get_storage_gateway(
    self,
    storage_gateway_id: str | UUID,
    *,
    include: str | Iterable[str] | None = None,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get storage gateway configuration.
    
    Parameters:
    - storage_gateway_id: UUID of the storage gateway
    - include: Additional data to include
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with gateway configuration
    """

def update_storage_gateway(
    self,
    storage_gateway_id: str | UUID,
    data: dict[str, Any] | StorageGatewayDocument,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Update storage gateway configuration.
    
    Modifies gateway settings including connector parameters,
    policies, and authentication requirements.
    
    Parameters:
    - storage_gateway_id: UUID of gateway to update
    - data: Updated configuration
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming update
    """

def delete_storage_gateway(
    self,
    storage_gateway_id: str | UUID,
    *,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Delete a storage gateway.
    
    Removes gateway configuration. Collections using this gateway
    will become inaccessible until configured with a new gateway.
    
    Parameters:
    - storage_gateway_id: UUID of gateway to delete
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming deletion
    """

Role and Access Management

Manage user roles and permissions for collections and endpoints with fine-grained access control.

def get_role_list(
    self,
    collection_id: str | UUID | None = None,
    include: str | None = None,
    query_params: dict[str, Any] | None = None
) -> IterableGCSResponse:
    """
    List roles with optional filtering by collection.
    
    Parameters:
    - collection_id: Filter roles for specific collection
    - include: Additional data to include
    - query_params: Additional parameters
    
    Returns:
    IterableGCSResponse with role listings
    """

def create_role(
    self,
    data: dict[str, Any] | GCSRoleDocument,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Create a user role assignment.
    
    Assigns specific permissions to a user or group for a collection,
    defining what operations they can perform (read, write, admin).
    
    Parameters:
    - data: Role assignment specification
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with created role details
    """

def get_role(
    self,
    role_id: str | UUID,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get role assignment details.
    
    Parameters:
    - role_id: UUID of the role assignment
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with role configuration
    """

def delete_role(
    self,
    role_id: str | UUID,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Remove a role assignment.
    
    Revokes user or group permissions for a collection.
    
    Parameters:
    - role_id: UUID of role assignment to delete
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming deletion
    """

User Credential Management

Manage user credentials for accessing storage systems that require authentication beyond Globus identity.

def get_user_credential_list(
    self,
    storage_gateway: str | UUID | None = None,
    query_params: dict[str, Any] | None = None
) -> IterableGCSResponse:
    """
    List user credentials with optional filtering.
    
    Parameters:
    - storage_gateway: Filter by associated storage gateway
    - query_params: Additional parameters
    
    Returns:
    IterableGCSResponse with credential listings
    """

def create_user_credential(
    self,
    data: dict[str, Any] | UserCredentialDocument,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Create user credential for storage access.
    
    Stores user credentials (username/password, keys, etc.) required
    for accessing underlying storage systems that need additional
    authentication beyond Globus identity.
    
    Parameters:
    - data: Credential specification including type and values
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with credential metadata (not sensitive data)
    """

def get_user_credential(
    self,
    user_credential_id: str | UUID,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Get user credential metadata.
    
    Returns credential information without exposing sensitive data.
    
    Parameters:
    - user_credential_id: UUID of the credential
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse with credential metadata
    """

def update_user_credential(
    self,
    user_credential_id: str | UUID,
    data: dict[str, Any] | UserCredentialDocument,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Update user credentials.
    
    Modifies stored credentials for storage system access.
    
    Parameters:
    - user_credential_id: UUID of credential to update
    - data: Updated credential information
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming update
    """

def delete_user_credential(
    self,
    user_credential_id: str | UUID,
    query_params: dict[str, Any] | None = None
) -> UnpackingGCSResponse:
    """
    Delete user credentials.
    
    Removes stored credentials, potentially making associated
    storage inaccessible until new credentials are provided.
    
    Parameters:
    - user_credential_id: UUID of credential to delete
    - query_params: Additional parameters
    
    Returns:
    UnpackingGCSResponse confirming deletion
    """

Configuration Documents and Data Classes

Comprehensive data structures for configuring endpoints, collections, storage gateways, and policies with type-safe construction.

class EndpointDocument(PayloadWrapper):
    """
    Document for configuring Globus Connect Server endpoints.
    
    Defines endpoint metadata including display name, description,
    organization information, contact details, and other settings.
    """
    
    def __init__(
        self,
        *,
        display_name: str | MissingType = MISSING,
        description: str | MissingType = MISSING,
        organization: str | MissingType = MISSING,
        department: str | MissingType = MISSING,
        contact_email: str | MissingType = MISSING,
        contact_info: str | MissingType = MISSING,
        info_link: str | MissingType = MISSING,
        public: bool | MissingType = MISSING,
        default_directory: str | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class CollectionDocument(PayloadWrapper):
    """
    Base class for collection configuration documents.
    
    Provides common functionality for both mapped and guest collections
    with policy configuration and access control.
    """
    
    def __init__(
        self,
        *,
        display_name: str | MissingType = MISSING,
        description: str | MissingType = MISSING,
        collection_type: str | MissingType = MISSING,
        public: bool | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class MappedCollectionDocument(CollectionDocument):
    """
    Configuration for mapped collections with direct storage access.
    
    Mapped collections provide direct access to storage through
    a storage gateway with configurable policies and permissions.
    """
    
    def __init__(
        self,
        *,
        storage_gateway_id: str | UUID | MissingType = MISSING,
        collection_base_path: str | MissingType = MISSING,
        policies: CollectionPolicies | dict | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class GuestCollectionDocument(CollectionDocument):
    """
    Configuration for guest collections with restricted access.
    
    Guest collections provide limited access to a subdirectory
    of a mapped collection with additional access controls.
    """
    
    def __init__(
        self,
        *,
        mapped_collection_id: str | UUID | MissingType = MISSING,
        root_path: str | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class StorageGatewayDocument(PayloadWrapper):
    """
    Configuration for storage gateways connecting to storage backends.
    
    Defines the connection between Globus services and underlying
    storage systems with connector-specific configuration and policies.
    """
    
    def __init__(
        self,
        *,
        display_name: str | MissingType = MISSING,
        connector: str | MissingType = MISSING,
        policies: StorageGatewayPolicies | dict | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class GCSRoleDocument(PayloadWrapper):
    """
    Document for role assignments and permissions management.
    
    Defines user or group permissions for collections including
    access levels and specific operation permissions.
    """
    
    def __init__(
        self,
        *,
        collection: str | UUID | MissingType = MISSING,
        principal: str | MissingType = MISSING,
        principal_type: str | MissingType = MISSING,
        role: str | MissingType = MISSING,
        **kwargs
    ) -> None: ...

class UserCredentialDocument(PayloadWrapper):
    """
    Document for user credential management.
    
    Stores user credentials required for accessing storage systems
    that need additional authentication beyond Globus identity.
    """
    
    def __init__(
        self,
        *,
        storage_gateway: str | UUID | MissingType = MISSING,
        identity_id: str | MissingType = MISSING,
        username: str | MissingType = MISSING,
        **kwargs
    ) -> None: ...

Storage Policies and Connectors

Comprehensive policy classes for different storage backend types with connector-specific configuration options.

class CollectionPolicies(PayloadWrapper):
    """Base class for collection-level policies and access controls."""

class POSIXCollectionPolicies(CollectionPolicies):
    """Policies for POSIX filesystem-based collections."""

class POSIXStagingCollectionPolicies(CollectionPolicies):
    """Policies for POSIX collections with staging support."""

class GoogleCloudStorageCollectionPolicies(CollectionPolicies):
    """Policies for Google Cloud Storage collections."""

class StorageGatewayPolicies(PayloadWrapper):
    """Base class for storage gateway policies and configuration."""

class POSIXStoragePolicies(StorageGatewayPolicies):
    """Policies for POSIX filesystem storage gateways."""

class S3StoragePolicies(StorageGatewayPolicies):
    """Policies for Amazon S3 storage gateways."""

class GoogleCloudStoragePolicies(StorageGatewayPolicies):
    """Policies for Google Cloud Storage gateways."""

class AzureBlobStoragePolicies(StorageGatewayPolicies):
    """Policies for Azure Blob Storage gateways."""

class BoxStoragePolicies(StorageGatewayPolicies):
    """Policies for Box storage gateways."""

class GoogleDriveStoragePolicies(StorageGatewayPolicies):
    """Policies for Google Drive storage gateways."""

class OneDriveStoragePolicies(StorageGatewayPolicies):
    """Policies for Microsoft OneDrive storage gateways."""

class ConnectorTable:
    """
    Registry and lookup table for storage connector information.
    
    Provides information about available storage connectors,
    their capabilities, and configuration requirements.
    """
    
    def __getitem__(self, connector_id: str) -> GlobusConnectServerConnector: ...
    
    def __contains__(self, connector_id: str) -> bool: ...

class GlobusConnectServerConnector:
    """
    Information about a specific storage connector.
    
    Contains metadata about connector capabilities, requirements,
    and configuration options for different storage backends.
    """

Response Objects

Specialized response classes for GCS operations with automatic unpacking and iteration support.

class IterableGCSResponse(IterableResponse):
    """Response class for GCS operations returning paginated data."""
    
    def __iter__(self) -> Iterator[dict[str, Any]]:
        """Iterate over GCS response data items."""

class UnpackingGCSResponse(GlobusHTTPResponse):
    """
    Response class that automatically unpacks GCS data.
    
    Provides direct access to the main data content from GCS
    responses while maintaining access to full response metadata.
    """

Error Handling

GCS-specific error handling for configuration and management operations.

class GCSAPIError(GlobusAPIError):
    """
    Error class for GCS Manager API errors.
    
    Provides enhanced error handling for GCS-specific error
    conditions including configuration validation and access issues.
    """

Common Usage Patterns

Basic Collection Setup

from globus_sdk import GCSClient, MappedCollectionDocument, POSIXStoragePolicies

# Initialize GCS client for your endpoint
gcs_client = GCSClient("gcs.example.org", authorizer=authorizer)

# Get endpoint information
endpoint_info = gcs_client.get_endpoint()
print(f"Endpoint: {endpoint_info['display_name']}")

# Create a storage gateway
storage_gateway_doc = StorageGatewayDocument(
    display_name="Main POSIX Storage",
    connector="posix",
    policies=POSIXStoragePolicies(
        groups_allow="0,users",
        groups_deny="",
        file_permissions="0644",
        directory_permissions="0755"
    )
)

gateway_response = gcs_client.create_storage_gateway(storage_gateway_doc)
gateway_id = gateway_response["id"]

# Create a mapped collection
collection_doc = MappedCollectionDocument(
    display_name="Research Data Collection",
    description="Shared research data access",
    storage_gateway_id=gateway_id,
    collection_base_path="/data/research",
    public=True
)

collection_response = gcs_client.create_collection(collection_doc)
collection_id = collection_response["id"]

print(f"Created collection: {collection_id}")

Role and Permission Management

from globus_sdk import GCSRoleDocument

# Grant read access to a user
read_role = GCSRoleDocument(
    collection=collection_id,
    principal="user@example.org",
    principal_type="identity",
    role="reader"
)

role_response = gcs_client.create_role(read_role)

# Grant write access to a group
write_role = GCSRoleDocument(
    collection=collection_id,
    principal="urn:globus:groups:id:12345678-1234-1234-1234-123456789012",
    principal_type="group",
    role="writer"
)

gcs_client.create_role(write_role)

# List all roles for the collection
roles = gcs_client.get_role_list(collection_id=collection_id)
for role in roles:
    print(f"Role {role['id']}: {role['principal']} -> {role['role']}")

Guest Collection Creation

from globus_sdk import GuestCollectionDocument

# Create a guest collection for limited access
guest_doc = GuestCollectionDocument(
    display_name="Project Alpha Data",
    description="Limited access to project alpha dataset",
    mapped_collection_id=collection_id,
    root_path="/project-alpha",
    public=False
)

guest_response = gcs_client.create_collection(guest_doc)
guest_collection_id = guest_response["id"]

# Grant access to specific users for the guest collection
guest_role = GCSRoleDocument(
    collection=guest_collection_id,
    principal="project-user@example.org", 
    principal_type="identity",
    role="writer"
)

gcs_client.create_role(guest_role)

User Credential Management

from globus_sdk import UserCredentialDocument

# Create user credentials for storage access
credential_doc = UserCredentialDocument(
    storage_gateway=gateway_id,
    identity_id="user-identity-uuid",
    username="storage_username"
    # Note: password/key would be provided through secure means
)

credential_response = gcs_client.create_user_credential(credential_doc)

# List credentials for monitoring
credentials = gcs_client.get_user_credential_list(storage_gateway=gateway_id)
for cred in credentials:
    print(f"Credential {cred['id']}: {cred['username']}")

Advanced Storage Configuration

from globus_sdk import S3StoragePolicies

# Create S3 storage gateway
s3_gateway_doc = StorageGatewayDocument(
    display_name="S3 Archive Storage",
    connector="s3",
    policies=S3StoragePolicies(
        s3_bucket="my-research-bucket",
        s3_region="us-west-2",
        s3_endpoint="https://s3.us-west-2.amazonaws.com"
    )
)

s3_gateway = gcs_client.create_storage_gateway(s3_gateway_doc)

# Create collection pointing to S3 storage
s3_collection_doc = MappedCollectionDocument(
    display_name="S3 Archive Collection",
    storage_gateway_id=s3_gateway["id"],
    collection_base_path="/",
    public=False
)

s3_collection = gcs_client.create_collection(s3_collection_doc)

Install with Tessl CLI