or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

auth-service.mdcompute-service.mdcore-framework.mdflows-service.mdgcs-service.mdgroups-service.mdindex.mdsearch-service.mdtimers-service.mdtransfer-service.md
tile.json

tessl/pypi-globus-sdk

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/globus-sdk@3.62.x

To install, run

npx @tessl/cli install tessl/pypi-globus-sdk@3.62.0

index.mddocs/

Globus SDK

A comprehensive Python SDK for the Globus platform, providing programmatic access to research data management services including data transfer, authentication, compute execution, workflow automation, and more. The Globus SDK enables seamless integration with the Globus ecosystem for scientific computing and collaboration workflows.

Package Information

  • Package Name: globus-sdk
  • Version: 3.62.0
  • Language: Python
  • Installation: pip install globus-sdk

Core Imports

import globus_sdk

Common service-specific imports:

from globus_sdk import (
    TransferClient, 
    AuthClient,
    ComputeClientV3,
    FlowsClient,
    GroupsClient,
    SearchClient,
    GCSClient,
    TimersClient
)

Basic Usage

Authentication and Authorization

The Globus SDK supports multiple authentication patterns using authorizers:

from globus_sdk import (
    AccessTokenAuthorizer, 
    TransferClient, 
    AuthClient,
    NativeAppAuthClient
)

# Using an access token directly
authorizer = AccessTokenAuthorizer("your_access_token")
client = TransferClient(authorizer=authorizer)

# OAuth2 flow for native applications
auth_client = NativeAppAuthClient("your_client_id")
auth_client.oauth2_start_flow()

# Get authorization URL for user
authorize_url = auth_client.oauth2_get_authorize_url()
print(f"Go to: {authorize_url}")

# Exchange authorization code for tokens
auth_code = input("Enter authorization code: ")
token_response = auth_client.oauth2_exchange_code_for_tokens(auth_code)

# Create authorizer with access token
transfer_token = token_response.by_resource_server['transfer.api.globus.org']
transfer_authorizer = AccessTokenAuthorizer(transfer_token['access_token'])
tc = TransferClient(authorizer=transfer_authorizer)

Data Transfer Operations

from globus_sdk import TransferClient, TransferData

# Initialize client
tc = TransferClient(authorizer=authorizer)

# Create a transfer operation
transfer_data = TransferData(
    source_endpoint="source_endpoint_id",
    destination_endpoint="dest_endpoint_id",
    label="My Transfer"
)

# Add items to transfer
transfer_data.add_item("/source/path/file.txt", "/dest/path/file.txt")
transfer_data.add_item("/source/dir/", "/dest/dir/", recursive=True)

# Submit the transfer
response = tc.submit_transfer(transfer_data)
task_id = response["task_id"]
print(f"Transfer submitted: {task_id}")

# Monitor transfer status
status = tc.get_task(task_id)
print(f"Status: {status['status']}")

Application Framework

from globus_sdk import UserApp, AuthClient, TransferClient

# Modern application-centric approach
app = UserApp(
    "my-app-name",
    client_id="your_client_id"
)

# Clients automatically use the app for authentication
auth_client = AuthClient(app=app)
transfer_client = TransferClient(app=app)

# The app manages tokens and scopes automatically

Architecture

The Globus SDK follows consistent design patterns across all services:

Service Client Pattern

  • All service clients inherit from BaseClient
  • Consistent initialization with app or authorizer parameters
  • Unified HTTP methods (get, post, put, patch, delete)
  • Service-specific error handling with APIError subclasses

Response Wrapper Pattern

  • All API responses wrapped in GlobusHTTPResponse or specialized subclasses
  • Dictionary-like access to response data
  • Consistent pagination support via IterableResponse
  • Rich response metadata (status codes, headers, etc.)

Authorization Strategy Pattern

  • Multiple authorizer implementations for different auth flows
  • Consistent GlobusAuthorizer interface
  • Support for OAuth2 flows, client credentials, access tokens, and auto-renewal

Scope Management System

  • Type-safe scope construction with service-specific ScopeBuilder objects
  • Dynamic scope resolution for collections and flows
  • Dependency handling for complex permission requirements

Capabilities

Authentication Service

OAuth2 flows, token management, identity resolution, and authorization for native and confidential applications. Supports authorization code, client credentials, and refresh token flows with comprehensive consent management.

class AuthClient(BaseClient):
    """General Auth API client for identity and authorization operations."""
    
    def __init__(
        self, 
        *, 
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def get_identities(
        self, 
        *, 
        usernames: list[str] | None = None,
        ids: list[str] | None = None,
        **params
    ) -> GetIdentitiesResponse: ...
    
    def get_identity_providers(self, **params) -> GlobusHTTPResponse: ...

class NativeAppAuthClient(AuthLoginClient):
    """OAuth client for native applications (public clients)."""
    
    def __init__(self, client_id: str, **kwargs) -> None: ...
    
    def oauth2_start_flow(
        self, 
        *, 
        requested_scopes: str | None = None,
        refresh_tokens: bool = False,
        prefill_named_grant: str | None = None
    ) -> None: ...
    
    def oauth2_get_authorize_url(self, **params) -> str: ...
    
    def oauth2_exchange_code_for_tokens(
        self, 
        auth_code: str
    ) -> OAuthTokenResponse: ...

class ConfidentialAppAuthClient(AuthLoginClient):
    """OAuth client for confidential applications (server-side apps)."""
    
    def __init__(
        self, 
        client_id: str, 
        client_secret: str, 
        **kwargs
    ) -> None: ...
    
    def oauth2_client_credentials_tokens(
        self, 
        *, 
        requested_scopes: str | None = None
    ) -> OAuthClientCredentialsResponse: ...

Authentication Service

Transfer Service

High-performance data movement between Globus endpoints with support for large files, directories, checksums, and advanced transfer options. Includes endpoint management, activation, and task monitoring.

class TransferClient(BaseClient):
    """Client for Globus Transfer API operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def submit_transfer(self, data: TransferData) -> GlobusHTTPResponse: ...
    
    def submit_delete(self, data: DeleteData) -> GlobusHTTPResponse: ...
    
    def get_task(self, task_id: str, **params) -> GlobusHTTPResponse: ...
    
    def task_list(
        self, 
        *, 
        limit: int | None = None,
        offset: int | None = None,
        **params
    ) -> IterableTransferResponse: ...
    
    def endpoint_search(
        self, 
        filter_fulltext: str | None = None, 
        *, 
        limit: int | None = None,
        offset: int | None = None,
        **params
    ) -> IterableTransferResponse: ...
    
    def get_endpoint(self, endpoint_id: str) -> GlobusHTTPResponse: ...

Transfer Service

Compute Service

Function execution and management on Globus Compute endpoints with support for Python functions, containers, and distributed computing patterns.

class ComputeClientV3(BaseClient):
    """Client for Globus Compute API v3."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def submit_function(
        self, 
        function_document: ComputeFunctionDocument
    ) -> GlobusHTTPResponse: ...
    
    def get_function(self, function_uuid: str) -> GlobusHTTPResponse: ...
    
    def submit_task(
        self, 
        endpoint_uuid: str, 
        function_uuid: str, 
        function_args: list | None = None,
        function_kwargs: dict | None = None
    ) -> GlobusHTTPResponse: ...
    
    def get_task(self, task_uuid: str) -> GlobusHTTPResponse: ...

Compute Service

Flows Service

Workflow automation and orchestration for complex multi-step operations across Globus services with conditional logic, error handling, and state management.

class FlowsClient(BaseClient):
    """Client for Globus Flows service operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def create_flow(
        self, 
        title: str, 
        definition: dict, 
        *, 
        input_schema: dict | None = None,
        **kwargs
    ) -> GlobusHTTPResponse: ...
    
    def get_flow(self, flow_id: str) -> GlobusHTTPResponse: ...
    
    def run_flow(
        self, 
        flow_id: str, 
        flow_input: dict, 
        *, 
        flow_scope: str | None = None,
        **kwargs
    ) -> GlobusHTTPResponse: ...

Flows Service

Globus Connect Server Management

Configuration and management of Globus Connect Server endpoints, collections, storage gateways, and access policies for institutional data sharing.

class GCSClient(BaseClient):
    """Client for Globus Connect Server management operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def create_endpoint(self, data: EndpointDocument) -> UnpackingGCSResponse: ...
    
    def get_endpoint(self, endpoint_id: str) -> UnpackingGCSResponse: ...
    
    def create_collection(self, data: CollectionDocument) -> UnpackingGCSResponse: ...
    
    def get_collection(self, collection_id: str) -> UnpackingGCSResponse: ...

GCS Service

Groups Service

Group membership management, access control, and policy configuration for collaborative research with fine-grained permission controls.

class GroupsClient(BaseClient):
    """Client for Globus Groups service operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def create_group(
        self, 
        name: str, 
        *, 
        description: str | None = None,
        **kwargs
    ) -> GlobusHTTPResponse: ...
    
    def get_group(self, group_id: str, **params) -> GlobusHTTPResponse: ...
    
    def add_member(
        self, 
        group_id: str, 
        identity_id: str, 
        *, 
        role: GroupRole = GroupRole.member
    ) -> GlobusHTTPResponse: ...

Groups Service

Search Service

Metadata indexing and search capabilities for research data discovery with support for custom schemas, faceted search, and real-time indexing.

class SearchClient(BaseClient):
    """Client for Globus Search service operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def search(
        self, 
        index_id: str, 
        q: str, 
        *, 
        limit: int | None = None,
        offset: int | None = None,
        **params
    ) -> GlobusHTTPResponse: ...
    
    def post_search(
        self, 
        index_id: str, 
        data: SearchQuery | dict
    ) -> GlobusHTTPResponse: ...

Search Service

Timer Service

Scheduled task execution and automation with support for one-time and recurring schedules, complex timing patterns, and integration with other Globus services.

class TimersClient(BaseClient):
    """Client for Globus Timers service operations."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None, 
        authorizer: GlobusAuthorizer | None = None,
        **kwargs
    ) -> None: ...
    
    def create_timer(
        self,
        timer: TimerJob | TransferTimer,
    ) -> GlobusHTTPResponse: ...
    
    def get_timer(self, timer_id: str) -> GlobusHTTPResponse: ...
    
    def list_timers(
        self, 
        *, 
        limit: int | None = None,
        **params
    ) -> GlobusHTTPResponse: ...

Timer Service

Core Framework

Foundation classes for authorization, HTTP clients, response handling, scope management, and error handling that provide consistent patterns across all Globus services.

class BaseClient:
    """Abstract base class for all Globus API clients."""
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        authorizer: GlobusAuthorizer | None = None,
        app_name: str | None = None,
        base_url: str | None = None,
        **kwargs
    ) -> None: ...
    
    def get(
        self, 
        path: str, 
        *, 
        query_params: dict[str, Any] | None = None,
        **kwargs
    ) -> GlobusHTTPResponse: ...
    
    def post(
        self, 
        path: str, 
        *, 
        data: Any = None,
        **kwargs
    ) -> GlobusHTTPResponse: ...

class GlobusHTTPResponse:
    """Response wrapper providing dict-like access to API response data."""
    
    @property
    def data(self) -> Any: ...
    
    @property 
    def http_status(self) -> int: ...
    
    def __getitem__(self, key: str) -> Any: ...
    
    def get(self, key: str, default: Any = None) -> Any: ...

Core Framework