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
transfer-service.mddocs/
Transfer Service
High-performance data movement between Globus endpoints with support for large files, directories, checksums, and advanced transfer options. The Transfer service provides reliable, resumable data transfer capabilities for research computing with comprehensive endpoint management and task monitoring.
Capabilities
Transfer Client
Core client for all Transfer service operations including data movement, endpoint management, and task monitoring with built-in support for pagination and filtering.
class TransferClient(BaseClient):
"""
Client for Globus Transfer API operations.
Provides methods for data transfer, endpoint management, task monitoring,
and collection operations with comprehensive filtering and pagination support.
"""
def __init__(
self,
*,
app: GlobusApp | None = None,
authorizer: GlobusAuthorizer | None = None,
environment: str | None = None,
base_url: str | None = None,
**kwargs
) -> None: ...
def add_app_data_access_scope(
self,
collection_ids: str | UUID | Iterable[str | UUID]
) -> TransferClient:
"""
Add dependent data_access scopes for collections to handle consent.
Used for resolving ConsentRequired errors when accessing
GCS Mapped Collections that require additional permissions.
Parameters:
- collection_ids: Collection ID(s) requiring data access scopes
Returns:
Self for method chaining
"""Data Transfer Operations
Submit and manage high-performance data transfers between Globus endpoints with comprehensive options for sync, verification, and encryption.
def submit_transfer(self, data: TransferData) -> GlobusHTTPResponse:
"""
Submit a transfer task for data movement between endpoints.
Parameters:
- data: TransferData object containing transfer specification
Returns:
GlobusHTTPResponse with task_id and submission details
"""
def submit_delete(self, data: DeleteData) -> GlobusHTTPResponse:
"""
Submit a delete task to remove files/directories from an endpoint.
Parameters:
- data: DeleteData object specifying files to delete
Returns:
GlobusHTTPResponse with task_id and submission details
"""
class TransferData(PayloadWrapper):
"""
Container for transfer request data with file/directory specifications.
Used to build transfer operations with items, sync options, and
advanced transfer parameters like encryption and checksumming.
"""
def __init__(
self,
transfer_client: TransferClient,
source_endpoint: str | UUID,
destination_endpoint: str | UUID,
*,
label: str | None = None,
submission_id: str | None = None,
sync_level: Literal["exists", "size", "mtime", "checksum"] | int = "size",
verify_checksum: bool = False,
preserve_timestamp: bool = False,
encrypt_data: bool = False,
deadline: str | datetime | None = None,
recursive_symlinks: Literal["ignore", "keep", "copy"] = "ignore",
skip_source_errors: bool = False,
fail_on_quota_errors: bool = False,
**kwargs
) -> None: ...
def add_item(
self,
source_path: str,
destination_path: str,
*,
recursive: bool = False,
external_checksum: str | None = None,
checksum_algorithm: str | None = None
) -> None:
"""
Add a file or directory to the transfer.
Parameters:
- source_path: Path on source endpoint
- destination_path: Path on destination endpoint
- recursive: Whether to recursively transfer directories
- external_checksum: Pre-computed checksum for verification
- checksum_algorithm: Algorithm used for checksum (md5, sha1, sha256, etc.)
"""
def add_symlink_item(
self,
source_path: str,
destination_path: str
) -> None:
"""Add a symbolic link to the transfer."""
class DeleteData(PayloadWrapper):
"""
Container for delete request data specifying files and directories to remove.
"""
def __init__(
self,
transfer_client: TransferClient,
endpoint: str | UUID,
*,
label: str | None = None,
submission_id: str | None = None,
recursive: bool = False,
deadline: str | datetime | None = None,
skip_source_errors: bool = False,
**kwargs
) -> None: ...
def add_item(self, path: str) -> None:
"""
Add a file or directory path to delete.
Parameters:
- path: Path to delete on the endpoint
"""Task Management
Monitor and control transfer tasks with comprehensive status tracking and task lifecycle management.
def get_task(
self,
task_id: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Get detailed information about a specific task.
Parameters:
- task_id: UUID of the task to retrieve
- query_params: Additional query parameters
Returns:
GlobusHTTPResponse with task details including status and statistics
"""
def task_list(
self,
*,
limit: int | None = None,
offset: int | None = None,
filter: str | dict | list[str | dict] | None = None,
**params
) -> IterableTransferResponse:
"""
List tasks with filtering and pagination support.
Parameters:
- limit: Maximum number of results to return
- offset: Offset for pagination
- filter: Task filter criteria (see filter formatting documentation)
- **params: Additional query parameters
Returns:
IterableTransferResponse with task listing
"""
def cancel_task(
self,
task_id: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Cancel a running or pending task.
Parameters:
- task_id: UUID of task to cancel
- query_params: Additional parameters
Returns:
GlobusHTTPResponse confirming cancellation
"""
def task_pause_info(
self,
task_id: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""Get information about why a task is paused."""
def task_resume(
self,
task_id: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""Resume a paused task."""
def task_wait(
self,
task_id: str,
*,
timeout: int = 10,
polling_interval: int = 10
) -> bool:
"""
Wait for a task to complete with configurable timeout and polling.
Parameters:
- task_id: UUID of task to wait for
- timeout: Maximum time to wait in seconds
- polling_interval: Seconds between status checks
Returns:
True if task completed successfully, False if timeout or failure
"""
def get_submission_id(self) -> GlobusHTTPResponse:
"""Get a submission ID for grouping related operations."""
def task_successful_transfers(
self,
task_id: str,
*,
limit: int | None = None,
**params
) -> IterableTransferResponse:
"""Get list of successfully transferred files from a task."""
def task_skipped_errors(
self,
task_id: str,
*,
limit: int | None = None,
**params
) -> IterableTransferResponse:
"""Get list of skipped errors from a transfer task."""Endpoint Management
Discover, retrieve, and manage Globus endpoints with activation support and detailed endpoint information.
def endpoint_search(
self,
filter_fulltext: str | None = None,
*,
limit: int | None = None,
offset: int | None = None,
filter: str | dict | list[str | dict] | None = None,
**params
) -> IterableTransferResponse:
"""
Search for endpoints using fulltext search and filters.
Parameters:
- filter_fulltext: Text search across endpoint names and descriptions
- limit: Maximum results to return
- offset: Pagination offset
- filter: Advanced filter criteria
- **params: Additional query parameters
Returns:
IterableTransferResponse with endpoint search results
"""
def get_endpoint(
self,
endpoint_id: str | UUID,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Get detailed information about a specific endpoint.
Parameters:
- endpoint_id: UUID of endpoint to retrieve
- query_params: Additional query parameters
Returns:
GlobusHTTPResponse with endpoint details
"""
def update_endpoint(
self,
endpoint_id: str | UUID,
data: dict[str, Any],
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Update endpoint configuration.
Parameters:
- endpoint_id: UUID of endpoint to update
- data: Partial endpoint document with updates
- query_params: Additional parameters
Returns:
GlobusHTTPResponse confirming update
"""
def endpoint_activate(
self,
endpoint_id: str | UUID,
*,
requirements_data: dict[str, Any] | None = None,
**params
) -> ActivationRequirementsResponse:
"""
Activate an endpoint for data operations.
Handles credential submission and activation workflows
required before endpoints can be used for transfers.
Parameters:
- endpoint_id: UUID of endpoint to activate
- requirements_data: Activation credentials/requirements
- **params: Additional activation parameters
Returns:
ActivationRequirementsResponse with activation status
"""
def endpoint_deactivate(
self,
endpoint_id: str | UUID,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""Deactivate an endpoint, clearing stored credentials."""
def endpoint_get_activation_requirements(
self,
endpoint_id: str | UUID,
*,
query_params: dict[str, Any] | None = None
) -> ActivationRequirementsResponse:
"""
Get activation requirements for an endpoint.
Returns information about what credentials or steps
are needed to activate the endpoint for use.
"""
def endpoint_autoactivate(
self,
endpoint_id: str | UUID,
*,
query_params: dict[str, Any] | None = None,
**kwargs
) -> GlobusHTTPResponse:
"""
Attempt automatic activation using stored credentials.
Returns:
GlobusHTTPResponse indicating success or additional requirements
"""File System Operations
Perform file system operations on Globus endpoints including directory listing, file operations, and path management.
def operation_ls(
self,
endpoint_id: str | UUID,
*,
path: str = "/",
show_hidden: bool = True,
limit: int | None = None,
offset: int | None = None,
**params
) -> IterableTransferResponse:
"""
List directory contents on an endpoint.
Parameters:
- endpoint_id: UUID of endpoint
- path: Directory path to list (default: "/")
- show_hidden: Include hidden files and directories
- limit: Maximum files to return
- offset: Pagination offset
Returns:
IterableTransferResponse with directory listing
"""
def operation_mkdir(
self,
endpoint_id: str | UUID,
path: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Create a directory on an endpoint.
Parameters:
- endpoint_id: UUID of endpoint
- path: Directory path to create
Returns:
GlobusHTTPResponse confirming directory creation
"""
def operation_rename(
self,
endpoint_id: str | UUID,
oldpath: str,
newpath: str,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""
Rename/move a file or directory on an endpoint.
Parameters:
- endpoint_id: UUID of endpoint
- oldpath: Current path
- newpath: New path
Returns:
GlobusHTTPResponse confirming rename operation
"""
def operation_stat(
self,
endpoint_id: str | UUID,
*,
path: str = "/",
**params
) -> GlobusHTTPResponse:
"""
Get file/directory metadata (stat information).
Parameters:
- endpoint_id: UUID of endpoint
- path: Path to get metadata for
Returns:
GlobusHTTPResponse with file metadata (size, type, permissions, etc.)
"""Collection Management
Manage collections on Globus Connect Server endpoints including guest collections and access policies.
def get_collection(
self,
collection_id: str | UUID,
*,
query_params: dict[str, Any] | None = None
) -> GlobusHTTPResponse:
"""Get detailed information about a collection."""
def collection_list(
self,
*,
mapped_collection_id: str | UUID | None = None,
limit: int | None = None,
**params
) -> IterableTransferResponse:
"""List collections with filtering options."""
def set_subscription_id(
self,
collection_id: str | UUID,
subscription_id: str | UUID | Literal["DEFAULT"] | None
) -> GlobusHTTPResponse:
"""
Set subscription ID on a mapped collection.
Used primarily for Globus Connect Personal subscription management.
"""Response Objects
Specialized response classes providing enhanced access to Transfer service data with pagination and activation support.
class IterableTransferResponse(IterableResponse):
"""
Response class for Transfer API operations that return paginated data.
Provides iteration support over result sets like task lists,
endpoint searches, and file listings.
"""
def __iter__(self) -> Iterator[dict[str, Any]]:
"""Iterate over response data items."""
class ActivationRequirementsResponse(GlobusHTTPResponse):
"""
Response class for endpoint activation operations.
Contains activation requirements, credentials needed,
and activation status information.
"""
@property
def active(self) -> bool:
"""Whether the endpoint is currently activated."""
@property
def auto_activation_supported(self) -> bool:
"""Whether the endpoint supports automatic activation."""
def get_requirement_value(
self,
requirement_name: str,
default: Any = None
) -> Any:
"""Get value for a specific activation requirement."""Error Handling
Transfer-specific error handling for common transfer scenarios and error conditions.
class TransferAPIError(GlobusAPIError):
"""
Error class for Transfer service API errors.
Provides enhanced error information for transfer-specific
error conditions like consent required, quota exceeded, etc.
"""Common Usage Patterns
Basic File Transfer
from globus_sdk import TransferClient, TransferData
# Initialize client
tc = TransferClient(authorizer=authorizer)
# Create transfer specification
transfer_data = TransferData(
tc,
source_endpoint="source-uuid",
destination_endpoint="dest-uuid",
label="My Transfer",
sync_level="checksum",
verify_checksum=True
)
# Add files/directories
transfer_data.add_item("/source/file.txt", "/dest/file.txt")
transfer_data.add_item("/source/dir/", "/dest/dir/", recursive=True)
# Submit and monitor
response = tc.submit_transfer(transfer_data)
task_id = response["task_id"]
# Wait for completion
tc.task_wait(task_id, timeout=3600)
task_info = tc.get_task(task_id)
print(f"Transfer status: {task_info['status']}")Endpoint Discovery and Activation
# Search for endpoints
endpoints = tc.endpoint_search("tutorial", limit=10)
for ep in endpoints:
print(f"{ep['display_name']}: {ep['id']}")
# Get endpoint details
endpoint = tc.get_endpoint("tutorial-endpoint-1")
print(f"Endpoint: {endpoint['display_name']}")
# Check activation requirements
req_response = tc.endpoint_get_activation_requirements("tutorial-endpoint-1")
if req_response.active:
print("Endpoint is already activated")
else:
# Try auto-activation
auto_response = tc.endpoint_autoactivate("tutorial-endpoint-1")
if auto_response["code"] == "AutoActivationFailed":
print("Manual activation required")
else:
print("Auto-activation successful")Directory Operations
# List directory contents
listing = tc.operation_ls("endpoint-uuid", path="/data", limit=100)
for item in listing:
print(f"{item['name']} ({'dir' if item['type'] == 'dir' else 'file'})")
# Create directory
tc.operation_mkdir("endpoint-uuid", "/data/new-directory")
# Get file metadata
file_info = tc.operation_stat("endpoint-uuid", path="/data/file.txt")
print(f"File size: {file_info['size']} bytes")Advanced Transfer Options
# High-assurance transfer with encryption and verification
transfer_data = TransferData(
tc,
source_endpoint="secure-endpoint",
destination_endpoint="backup-endpoint",
label="Secure Backup",
encrypt_data=True,
verify_checksum=True,
preserve_timestamp=True,
deadline="2024-12-31T23:59:59Z"
)
# Add files with pre-computed checksums
transfer_data.add_item(
"/data/important.dat",
"/backup/important.dat",
external_checksum="sha256:abc123...",
checksum_algorithm="sha256"
)
# Submit with error handling
try:
response = tc.submit_transfer(transfer_data)
print(f"Transfer submitted: {response['task_id']}")
except TransferAPIError as e:
if "ConsentRequired" in str(e):
print("Additional consent required for collections")
else:
print(f"Transfer failed: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-globus-sdk