Client Operations

class Client(BaseClient):
    """
    Main client class handling Google Directory API interactions.
    
    Extends Airbyte CDK BaseClient to provide Google Directory-specific
    functionality including authentication, health checks, and stream management.
    """
    
    def __init__(self, credentials: Mapping[str, Any] = None, credentials_json: str = None, email: str = None):
        """
        Initialize the Google Directory client.
        
        Parameters:
        - credentials: Dict containing authentication credentials
        - credentials_json: JSON string of service account credentials (legacy)
        - email: Admin email for service account delegation (legacy)
        
        The constructor supports both new (credentials dict) and legacy
        (credentials_json + email) configuration formats.
        """

def health_check(self) -> Tuple[bool, str]:
    """
    Perform health check against Google Directory API.
    
    Tests connectivity by attempting to fetch users from the directory.
    
    Returns:
    - Tuple[bool, str]: (is_healthy, error_message)
        - is_healthy: True if connection successful, False otherwise  
        - error_message: None if healthy, error description if unhealthy
    """

@property
def streams(self) -> Generator[AirbyteStream, None, None]:
    """
    Generator yielding AirbyteStream objects with primary keys configured.
    
    All streams are configured with source_defined_primary_key = [["id"]]
    to use the 'id' field as the primary key for change detection.
    
    Yields:
    - AirbyteStream objects for users, groups, and group_members streams
    """

def _enumerate_methods(self) -> Mapping[str, callable]:
    """
    Enumerate available stream methods for the Airbyte CDK.
    
    Maps stream names to their corresponding API list methods from
    the configured stream APIs (users, groups, group_members).
    
    Returns:
    - Mapping[str, callable]: Dictionary mapping stream names to list methods
    """

def get_stream_state(self, name: str) -> Any:
    """
    Get stream state for incremental synchronization.
    
    Parameters:
    - name: Stream name
    
    Returns:
    - Any: Stream state (currently not implemented)
    """

def set_stream_state(self, name: str, state: Any):
    """
    Set stream state for incremental synchronization.
    
    Parameters:
    - name: Stream name
    - state: State to persist
    
    Currently not implemented - state management is handled by Airbyte platform.
    """

from source_google_directory.client import Client

# OAuth credentials configuration
oauth_credentials = {
    "client_id": "your-oauth-client-id",
    "client_secret": "your-oauth-client-secret",
    "refresh_token": "your-refresh-token"
}

# Initialize client
client = Client(credentials=oauth_credentials)

# Perform health check
is_healthy, error_msg = client.health_check()
if is_healthy:
    print("Connection successful!")
else:
    print(f"Connection failed: {error_msg}")

from source_google_directory.client import Client

# Service account credentials
service_credentials = {
    "credentials_json": '{"type": "service_account", "project_id": "...", ...}',
    "email": "admin@yourdomain.com"
}

# Initialize client
client = Client(credentials=service_credentials)

# Check connection health
is_healthy, error_msg = client.health_check()

# Legacy format (still supported)
client = Client(
    credentials_json='{"type": "service_account", ...}',
    email="admin@yourdomain.com"
)

# Get available streams
for stream in client.streams:
    print(f"Stream: {stream.name}")
    print(f"Primary key: {stream.source_defined_primary_key}")