Core Framework

class BaseClient:
    """
    Abstract base class for all Globus API clients.
    
    Provides consistent HTTP operations, authentication handling, error processing,
    and response wrapping across all Globus service clients with standardized
    initialization patterns and request/response lifecycle management.
    """
    
    def __init__(
        self,
        *,
        app: GlobusApp | None = None,
        app_scopes: list[Scope] | None = None,
        authorizer: GlobusAuthorizer | None = None,
        app_name: str | None = None,
        base_url: str | None = None,
        environment: str | None = None,
        transport_params: dict[str, Any] | None = None
    ) -> None: ...
    
    @property
    def default_scope_requirements(self) -> list[Scope]:
        """
        Default scopes required for client operations.
        
        Returns the scopes that will automatically be added to the
        client's app scope requirements during initialization.
        """
    
    @property
    def resource_server(self) -> str | None:
        """
        Resource server identifier for this client.
        
        Returns the resource server name associated with this client's
        scopes and authentication requirements.
        """
    
    @property
    def app_name(self) -> str | None:
        """Application name for User-Agent headers."""
    
    def attach_globus_app(
        self, 
        app: GlobusApp, 
        app_scopes: list[Scope] | None = None
    ) -> None:
        """
        Attach a GlobusApp to this client.
        
        Registers the client with the app and adds scope requirements
        for automatic authentication and token management.
        
        Parameters:
        - app: GlobusApp instance to attach
        - app_scopes: Optional custom scopes instead of defaults
        """
    
    def add_app_scope(
        self, 
        scope_collection: ScopeCollectionType
    ) -> BaseClient:
        """
        Add additional scopes to the client's app requirements.
        
        Extends the scope requirements beyond default scopes for
        accessing additional resources or permissions.
        
        Parameters:
        - scope_collection: Scopes to add to requirements
        
        Returns:
        Self for method chaining
        """
    
    def get(
        self,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """
        Make a GET request to the specified path.
        
        Parameters:
        - path: Request path (relative to base_url)
        - query_params: URL query parameters
        - headers: Additional HTTP headers
        - automatic_authorization: Use client's authorizer
        
        Returns:
        GlobusHTTPResponse with response data and metadata
        """
    
    def post(
        self,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        data: Any = None,
        headers: dict[str, str] | None = None,
        encoding: str | None = None,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """
        Make a POST request to the specified path.
        
        Parameters:
        - path: Request path
        - query_params: URL query parameters
        - data: Request body data
        - headers: Additional headers
        - encoding: Data encoding (json, form, text)
        - automatic_authorization: Use client's authorizer
        
        Returns:
        GlobusHTTPResponse with response data
        """
    
    def put(
        self,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        data: Any = None,
        headers: dict[str, str] | None = None,
        encoding: str | None = None,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """Make a PUT request to the specified path."""
    
    def patch(
        self,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        data: Any = None,
        headers: dict[str, str] | None = None,
        encoding: str | None = None,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """Make a PATCH request to the specified path."""
    
    def delete(
        self,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """Make a DELETE request to the specified path."""
    
    def request(
        self,
        method: str,
        path: str,
        *,
        query_params: dict[str, Any] | None = None,
        data: Any = None,
        headers: dict[str, str] | None = None,
        encoding: str | None = None,
        allow_redirects: bool = True,
        stream: bool = False,
        automatic_authorization: bool = True
    ) -> GlobusHTTPResponse:
        """
        Send an HTTP request with full parameter control.
        
        Provides complete control over HTTP request parameters with
        consistent error handling and response processing.
        
        Parameters:
        - method: HTTP method (GET, POST, PUT, PATCH, DELETE)
        - path: Request path
        - query_params: URL query parameters
        - data: Request body data
        - headers: HTTP headers
        - encoding: Data encoding method
        - allow_redirects: Follow redirects automatically
        - stream: Stream response without immediate download
        - automatic_authorization: Use client authentication
        
        Returns:
        GlobusHTTPResponse wrapping the response
        
        Raises:
        GlobusAPIError for 4xx/5xx HTTP status codes
        """

class GlobusAuthorizer(metaclass=ABCMeta):
    """
    Abstract base class for all authorization handlers.
    
    Defines the interface for generating Authorization headers and
    handling authorization failures with consistent patterns across
    different authentication methods.
    """
    
    @abstractmethod
    def get_authorization_header(self) -> str | None:
        """
        Generate Authorization header value.
        
        Returns the complete Authorization header value to be included
        in HTTP requests, or None if no authorization should be used.
        
        Returns:
        Authorization header string or None
        """
    
    def handle_missing_authorization(self) -> bool:
        """
        Handle 401 Unauthorized responses.
        
        Called when a request returns 401 status to allow the authorizer
        to attempt recovery (e.g., token refresh). Returns True if the
        authorizer believes it has fixed the issue.
        
        Returns:
        True if authorization was updated, False otherwise
        """

class AccessTokenAuthorizer(StaticGlobusAuthorizer):
    """
    Authorization using a single access token.
    
    Provides Bearer token authorization for requests using a static
    access token without refresh capabilities.
    """
    
    def __init__(self, access_token: str) -> None: ...
    
    @property
    def access_token(self) -> str:
        """The access token used for authorization."""
    
    @property  
    def access_token_hash(self) -> str:
        """SHA256 hash of the access token for logging."""

class RefreshTokenAuthorizer(GlobusAuthorizer):
    """
    Authorization using refresh tokens for automatic token renewal.
    
    Automatically refreshes access tokens when they expire using
    stored refresh tokens with configurable retry and caching.
    """
    
    def __init__(
        self,
        refresh_token: str,
        auth_client: AuthClient,
        access_token: str | None = None,
        expires_at: int | None = None,
        on_refresh: Callable[[OAuthTokenResponse], None] | None = None
    ) -> None: ...
    
    def ensure_valid_token(self) -> str:
        """Ensure access token is valid, refreshing if necessary."""

class ClientCredentialsAuthorizer(GlobusAuthorizer):
    """
    Authorization using OAuth2 Client Credentials flow.
    
    Automatically obtains and refreshes access tokens using client
    credentials (client ID and secret) for service-to-service authentication.
    """
    
    def __init__(
        self,
        confidential_client: ConfidentialAppAuthClient,
        scopes: str | Iterable[str] | None = None,
        access_token: str | None = None,
        expires_at: int | None = None
    ) -> None: ...

class BasicAuthorizer(StaticGlobusAuthorizer):
    """
    HTTP Basic authentication authorizer.
    
    Provides HTTP Basic authentication using username and password
    with base64 encoding according to RFC 7617.
    """
    
    def __init__(self, username: str, password: str) -> None: ...

class NullAuthorizer(GlobusAuthorizer):
    """
    No-authentication authorizer.
    
    Ensures no Authorization header is sent with requests, useful
    for public APIs or when authentication is handled elsewhere.
    """
    
    def get_authorization_header(self) -> None:
        """Always returns None (no authorization)."""

class GlobusHTTPResponse:
    """
    Response wrapper providing dict-like access to API response data.
    
    Wraps HTTP responses with convenient data access methods, error handling,
    and metadata preservation while maintaining access to raw response data.
    """
    
    def __init__(
        self,
        response: Response | GlobusHTTPResponse,
        client: BaseClient | None = None
    ) -> None: ...
    
    @property
    def data(self) -> Any:
        """
        Parsed JSON response data.
        
        Returns the JSON-parsed response body, or None if the response
        was not valid JSON or contained no data.
        """
    
    @property
    def text(self) -> str:
        """Raw response text content."""
    
    @property
    def binary_content(self) -> bytes:
        """Raw response binary content."""
    
    @property
    def http_status(self) -> int:
        """HTTP status code from the response."""
    
    @property
    def http_reason(self) -> str:
        """HTTP reason phrase from the response."""
    
    @property
    def headers(self) -> dict[str, str]:
        """HTTP response headers as a dictionary."""
    
    @property
    def content_type(self) -> str | None:
        """Content-Type header value."""
    
    def __getitem__(self, key: str) -> Any:
        """
        Dictionary-style access to response data.
        
        Provides convenient access to JSON response fields using
        bracket notation for dict-like usage patterns.
        """
    
    def __contains__(self, key: str) -> bool:
        """Check if key exists in response data."""
    
    def __bool__(self) -> bool:
        """Response truthiness based on HTTP success."""
    
    def get(self, key: str, default: Any = None) -> Any:
        """
        Get response data field with default fallback.
        
        Parameters:
        - key: Field name to retrieve
        - default: Default value if key not found
        
        Returns:
        Field value or default
        """

class IterableResponse(GlobusHTTPResponse):
    """
    Response class for paginated data with iteration support.
    
    Provides iteration over response data arrays with automatic
    key resolution and consistent iteration patterns across services.
    """
    
    @property
    def default_iter_key(self) -> str | None:
        """Default key name for iteration data."""
    
    @property
    def iter_key(self) -> str:
        """Key name used for iteration."""
    
    def __iter__(self) -> Iterator[Any]:
        """
        Iterate over response data items.
        
        Automatically resolves the appropriate data array from the
        response and provides iteration over individual items.
        """

class ArrayResponse(GlobusHTTPResponse):
    """
    Response class for JSON array data with enhanced array operations.
    
    Specialized for responses that contain JSON arrays at the top level
    with length operations and efficient iteration.
    """
    
    def __iter__(self) -> Iterator[Any]:
        """Iterate over array elements."""
    
    def __len__(self) -> int:
        """Get number of elements in the array."""

class Scope:
    """
    Representation of a Globus scope with dependency management.
    
    Represents a single scope string with support for dependent scopes,
    optional scopes, and scope relationships for complex permission structures.
    """
    
    def __init__(
        self,
        scope: str,
        *,
        optional: bool = False,
        dependencies: Iterable[Scope] | None = None
    ) -> None: ...
    
    @property
    def scope_string(self) -> str:
        """The scope string value."""
    
    @property
    def optional(self) -> bool:
        """Whether this scope is optional."""
    
    @property
    def dependencies(self) -> list[Scope]:
        """List of dependent scopes."""
    
    def add_dependency(self, scope: Scope) -> None:
        """
        Add a dependent scope.
        
        Creates a dependency relationship where this scope requires
        the dependent scope to be granted as well.
        
        Parameters:
        - scope: Scope that this scope depends on
        """
    
    def serialize(self) -> str:
        """
        Serialize scope and dependencies to string format.
        
        Converts the scope and its dependency tree to the string
        representation used in OAuth2 flows and API requests.
        
        Returns:
        Serialized scope string with dependencies
        """

class ScopeBuilder:
    """
    Base class for service-specific scope builders.
    
    Provides a foundation for constructing scopes with service-specific
    patterns and relationships while maintaining type safety.
    """
    
    @property
    def resource_server(self) -> str:
        """Resource server identifier for this scope builder."""

class MutableScope:
    """
    Mutable scope object for dynamic scope construction.
    
    Allows modification of scope properties and dependencies after
    creation for complex scope building scenarios.
    """
    
    def __init__(
        self,
        scope_string: str,
        *,
        optional: bool = False,
        dependencies: list[MutableScope] | None = None
    ) -> None: ...
    
    def add_dependency(self, scope: MutableScope) -> None:
        """Add a dependency to this scope."""
    
    def freeze(self) -> Scope:
        """
        Convert to immutable Scope object.
        
        Freezes the mutable scope into an immutable Scope instance
        with all current dependencies and properties preserved.
        
        Returns:
        Immutable Scope representation
        """

# Service-specific scope builders
class AuthScopes(ScopeBuilder):
    """Scope builder for Auth service scopes."""
    
    openid: str = "openid"
    profile: str = "profile" 
    email: str = "email"
    view_identity_set: str = "urn:globus:auth:scope:auth.globus.org:view_identity_set"

class TransferScopes(ScopeBuilder):
    """Scope builder for Transfer service scopes."""
    
    all: str = "urn:globus:auth:scope:transfer.api.globus.org:all"

class ComputeScopes(ScopeBuilder):
    """Scope builder for Compute service scopes."""
    
    all: str = "urn:globus:auth:scope:compute.api.globus.org:all"

class FlowsScopes(ScopeBuilder):
    """Scope builder for Flows service scopes."""
    
    all: str = "urn:globus:auth:scope:flows.api.globus.org:all"
    manage_flows: str = "urn:globus:auth:scope:flows.api.globus.org:manage_flows"
    run_manage: str = "urn:globus:auth:scope:flows.api.globus.org:run_manage"

# Dynamic scope builders for resource-specific scopes
class GCSCollectionScopeBuilder(ScopeBuilder):
    """Dynamic scope builder for GCS collection-specific scopes."""
    
    def __init__(self, collection_id: str) -> None: ...
    
    @property
    def data_access(self) -> str:
        """Data access scope for this collection."""

class GCSEndpointScopeBuilder(ScopeBuilder):
    """Dynamic scope builder for GCS endpoint-specific scopes."""
    
    def __init__(self, endpoint_id: str) -> None: ...
    
    @property
    def manage_collections(self) -> str:
        """Collection management scope for this endpoint."""

# Utility functions
def scopes_to_str(scopes: ScopeCollectionType) -> str:
    """
    Convert scope collection to space-separated string.
    
    Parameters:
    - scopes: Collection of scopes to convert
    
    Returns:
    Space-separated scope string for OAuth2 flows
    """

def scopes_to_scope_list(scopes: ScopeCollectionType) -> list[Scope]:
    """
    Normalize scope collection to list of Scope objects.
    
    Parameters:
    - scopes: Mixed collection of scopes to normalize
    
    Returns:
    Normalized list of Scope objects
    """

class GlobusError(Exception):
    """
    Base exception class for all Globus SDK errors.
    
    Root of the Globus SDK exception hierarchy providing common
    error handling patterns and metadata preservation.
    """

class GlobusSDKUsageError(GlobusError):
    """
    Exception for SDK misuse and configuration errors.
    
    Raised when the SDK is used incorrectly, such as invalid
    parameter combinations or missing required configuration.
    """

class ValidationError(GlobusSDKUsageError):
    """
    Exception for input validation failures.
    
    Raised when user-provided data fails validation checks
    before being sent to Globus services.
    """

class GlobusAPIError(GlobusError):
    """
    Base exception for HTTP 4xx and 5xx API errors.
    
    Provides access to HTTP response data, error details, and
    structured error information from Globus services.
    """
    
    def __init__(self, response: Response) -> None: ...
    
    @property
    def http_status(self) -> int:
        """HTTP status code from the error response."""
    
    @property
    def http_reason(self) -> str:
        """HTTP reason phrase from the error response."""
    
    @property
    def headers(self) -> dict[str, str]:
        """HTTP headers from the error response."""
    
    @property
    def text(self) -> str:
        """Raw error response text."""
    
    @property
    def binary_content(self) -> bytes:
        """Raw error response binary content."""
    
    @property
    def data(self) -> Any:
        """Parsed JSON error data if available."""
    
    @property
    def code(self) -> str:
        """Globus error code from the response."""
    
    @property
    def message(self) -> str:
        """Human-readable error message."""
    
    def __getitem__(self, key: str) -> Any:
        """Dictionary-style access to error data."""
    
    def __contains__(self, key: str) -> bool:
        """Check if key exists in error data."""
    
    def get(self, key: str, default: Any = None) -> Any:
        """Get error data field with default."""

class ErrorSubdocument:
    """
    Container for detailed error information from API responses.
    
    Represents individual error details within structured error
    responses with code, message, and additional metadata.
    """
    
    def __init__(self, data: dict[str, Any]) -> None: ...
    
    @property
    def code(self) -> str:
        """Error code identifier."""
    
    @property
    def message(self) -> str:
        """Human-readable error message."""
    
    @property
    def detail(self) -> str:
        """Detailed error description."""

# Network-related errors
class NetworkError(GlobusError):
    """Base class for network connectivity errors."""

class GlobusTimeoutError(NetworkError):
    """Request timeout errors."""

class GlobusConnectionError(NetworkError):
    """Connection establishment errors."""

class GlobusConnectionTimeoutError(GlobusConnectionError):
    """Connection timeout errors."""

# Error information containers
class ErrorInfo:
    """
    Container for structured error information.
    
    Provides access to error details, codes, and additional
    context information from API error responses.
    """

class ErrorInfoContainer:
    """
    Collection of ErrorInfo objects.
    
    Manages multiple error information objects with iteration
    and lookup capabilities for complex error scenarios.
    """

class ConsentRequiredInfo(ErrorInfo):
    """
    Specialized error info for consent required errors.
    
    Provides structured information about required consents
    and authorization parameters for resolving access issues.
    """
    
    @property
    def required_scopes(self) -> list[str]:
        """Scopes requiring user consent."""
    
    @property
    def authorization_parameters(self) -> dict[str, Any]:
        """OAuth2 parameters for consent flow."""

# Warnings
class RemovedInV4Warning(UserWarning):
    """
    Warning for features that will be removed in SDK v4.0.
    
    Indicates deprecated functionality that should be migrated
    to newer alternatives before the next major version.
    """

class PayloadWrapper:
    """
    Base class for API request payload helpers.
    
    Provides a foundation for building structured request data with
    validation, serialization, and convenience methods for API operations.
    """
    
    def __init__(self) -> None: ...
    
    def __getitem__(self, key: str) -> Any:
        """Dictionary-style access to payload data."""
    
    def __setitem__(self, key: str, value: Any) -> None:
        """Dictionary-style assignment to payload data."""
    
    def __contains__(self, key: str) -> bool:
        """Check if key exists in payload data."""
    
    def get(self, key: str, default: Any = None) -> Any:
        """Get payload field with default fallback."""

class MissingType:
    """
    Sentinel type for distinguishing missing values from None.
    
    Used throughout the SDK to represent truly missing/unset values
    as distinct from explicit None values.
    """

# Sentinel instance
MISSING: MissingType = MissingType()

# Utility functions
def filter_missing(data: dict[str, Any]) -> dict[str, Any]:
    """
    Remove MISSING values from dictionary.
    
    Filters out all keys with MISSING sentinel values to produce
    clean dictionaries for API requests.
    
    Parameters:
    - data: Dictionary potentially containing MISSING values
    
    Returns:
    Dictionary with MISSING values removed
    """

def slash_join(*parts: str) -> str:
    """
    Join URL path components with proper slash handling.
    
    Joins path segments ensuring single slashes between components
    and proper handling of leading/trailing slashes.
    
    Parameters:
    - *parts: Path components to join
    
    Returns:
    Properly formatted URL path
    """

def sha256_string(data: str) -> str:
    """
    Compute SHA256 hash of a string.
    
    Parameters:
    - data: String to hash
    
    Returns:
    Hexadecimal SHA256 hash
    """

def b64str(data: str) -> str:
    """
    Base64 encode a string.
    
    Parameters:
    - data: String to encode
    
    Returns:
    Base64 encoded string
    """

from globus_sdk import TransferClient, AccessTokenAuthorizer

# Simple token-based authorization
authorizer = AccessTokenAuthorizer("your_access_token")
client = TransferClient(authorizer=authorizer)

# Application-based initialization  
from globus_sdk import UserApp

app = UserApp("my-app", client_id="your_client_id")
client = TransferClient(app=app)

from globus_sdk import RefreshTokenAuthorizer, AuthClient

# Refresh token authorization with callback
def token_refresh_callback(token_response):
    # Save new tokens to persistent storage
    save_tokens(token_response.by_resource_server)

auth_client = AuthClient()
authorizer = RefreshTokenAuthorizer(
    refresh_token="your_refresh_token",
    auth_client=auth_client,
    on_refresh=token_refresh_callback
)

client = TransferClient(authorizer=authorizer)

from globus_sdk import TransferScopes, ComputeScopes, Scope

# Using service scope builders
base_scopes = [
    Scope(TransferScopes.all),
    Scope(ComputeScopes.all)
]

# Dynamic collection scopes
from globus_sdk import GCSCollectionScopeBuilder

collection_scope_builder = GCSCollectionScopeBuilder("collection-uuid")
data_access_scope = Scope(collection_scope_builder.data_access, optional=True)

# Scope with dependencies
transfer_scope = Scope(TransferScopes.all)
transfer_scope.add_dependency(data_access_scope)

from globus_sdk import GlobusAPIError, ConsentRequiredInfo

try:
    response = client.submit_transfer(transfer_data)
except GlobusAPIError as e:
    if e.http_status == 403:
        # Check for consent required errors
        if e.info.consent_required:
            consent_info = ConsentRequiredInfo(e)
            print(f"Consent required for scopes: {consent_info.required_scopes}")
            
            # Handle consent flow
            auth_params = consent_info.authorization_parameters
            # Redirect user to consent URL...
    else:
        print(f"API Error {e.http_status}: {e.message}")
        # Handle other API errors
except Exception as e:
    print(f"Unexpected error: {e}")

# Dictionary-style access
response = client.get_endpoint("endpoint-uuid")
print(f"Endpoint name: {response['display_name']}")

# Iterate over paginated results
endpoints = client.endpoint_search("tutorial")
for endpoint in endpoints:
    print(f"Found: {endpoint['display_name']}")

# Access response metadata
print(f"Status: {response.http_status}")
print(f"Headers: {response.headers}")

from globus_sdk import BaseClient

class CustomClient(BaseClient):
    service_name = "my_service"
    error_class = MyCustomAPIError
    
    @property
    def default_scope_requirements(self):
        return [Scope("https://my-service.org/scope")]
    
    def custom_operation(self, data):
        return self.post("/custom", data=data)

# Use custom transport parameters
custom_client = CustomClient(
    authorizer=authorizer,
    transport_params={
        "timeout": 60,
        "retries": 3,
        "user_agent_suffix": "MyApp/1.0"
    }
)