tessl/pypi-quilt3

Quilt manages data like code with packages, repositories, browsing and revision history for machine learning and data-driven domains

—

Pending

Overview

Eval results

Files

Hooks Module

Name: tessl/pypi-quilt3
Author: tessl

Extension system for customizing Quilt3 behavior through configurable hook functions. Allows users to modify or extend core functionality by providing custom implementations for specific operations.

Type Imports

from typing import Optional, Callable

Capabilities

S3 Client Hook Management

Configure custom S3 client builders for specialized AWS operations.

def get_build_s3_client_hook() -> Optional[BuildClientHook]:
    """
    Return the current build S3 client hook.

    Returns:
    The currently configured hook function, or None if no hook is set
    """

def set_build_s3_client_hook(hook: Optional[BuildClientHook]) -> Optional[BuildClientHook]:
    """
    Set the build S3 client hook.

    Parameters:
    - hook: Hook function to set, or None to clear the current hook

    Returns:
    The previously configured hook function, or None if no hook was set
    """

Hook Types

BuildClientHook

BuildClientHook = Callable[..., Any]
"""
Type alias for S3 client builder hook functions.

Hook functions should accept boto3 session parameters and return
a configured S3 client or modify client behavior.
"""

Usage Examples

Basic Hook Configuration

import quilt3.hooks

# Check current hook configuration
current_hook = quilt3.hooks.get_build_s3_client_hook()
if current_hook:
    print("S3 client hook is currently configured")
else:
    print("No S3 client hook configured")

# Clear any existing hook
previous_hook = quilt3.hooks.set_build_s3_client_hook(None)
print("Cleared S3 client hook")

Custom S3 Client Hook

import quilt3.hooks
import boto3

def custom_s3_client_builder(*args, **kwargs):
    """
    Custom S3 client builder with additional configuration.
    
    This example adds custom retry configuration and timeout settings.
    """
    # Create base session
    session = boto3.Session(*args, **kwargs)
    
    # Configure custom S3 client with retries and timeouts
    s3_client = session.client(
        's3',
        config=boto3.session.Config(
            retries={'max_attempts': 10, 'mode': 'adaptive'},
            connect_timeout=60,
            read_timeout=60
        )
    )
    
    return s3_client

# Install the custom hook
previous_hook = quilt3.hooks.set_build_s3_client_hook(custom_s3_client_builder)

# Now all Quilt operations will use the custom S3 client
import quilt3
bucket = quilt3.Bucket("s3://my-bucket")  # Will use custom client configuration

Regional S3 Client Hook

import quilt3.hooks
import boto3

def regional_s3_client_builder(*args, **kwargs):
    """
    Custom S3 client builder that forces a specific region.
    """
    # Extract region from kwargs or use default
    region = kwargs.pop('region_name', 'us-west-2')
    
    session = boto3.Session(*args, **kwargs)
    
    # Force specific region for all S3 operations
    s3_client = session.client('s3', region_name=region)
    
    print(f"Created S3 client for region: {region}")
    return s3_client

# Install regional hook
quilt3.hooks.set_build_s3_client_hook(regional_s3_client_builder)

# Test the hook
import quilt3
bucket = quilt3.Bucket("s3://my-bucket")
bucket.ls()  # Will use forced region configuration

Development vs Production Hooks

import quilt3.hooks
import boto3
import os

def environment_aware_s3_client(*args, **kwargs):
    """
    S3 client builder that adapts to development vs production environments.
    """
    env = os.getenv('ENVIRONMENT', 'development')
    
    session = boto3.Session(*args, **kwargs)
    
    if env == 'development':
        # Development: use localstack or minio
        s3_client = session.client(
            's3',
            endpoint_url='http://localhost:4566',  # LocalStack
            aws_access_key_id='test',
            aws_secret_access_key='test',
            region_name='us-east-1'
        )
        print("Using development S3 endpoint")
    else:
        # Production: use standard AWS S3
        s3_client = session.client('s3')
        print("Using production AWS S3")
    
    return s3_client

# Configure environment-aware hook
quilt3.hooks.set_build_s3_client_hook(environment_aware_s3_client)

Hook with Logging and Monitoring

import quilt3.hooks
import boto3
import logging
import time

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitored_s3_client_builder(*args, **kwargs):
    """
    S3 client builder with logging and performance monitoring.
    """
    start_time = time.time()
    
    try:
        session = boto3.Session(*args, **kwargs)
        s3_client = session.client('s3')
        
        build_time = time.time() - start_time
        logger.info(f"S3 client built successfully in {build_time:.3f}s")
        
        return s3_client
        
    except Exception as e:
        build_time = time.time() - start_time
        logger.error(f"S3 client build failed after {build_time:.3f}s: {e}")
        raise

# Install monitoring hook
quilt3.hooks.set_build_s3_client_hook(monitored_s3_client_builder)

# All S3 operations will now be logged
import quilt3
bucket = quilt3.Bucket("s3://my-bucket")
files = bucket.ls()  # This will log client creation

Temporary Hook Usage

import quilt3.hooks
import contextlib

@contextlib.contextmanager
def temporary_s3_hook(hook_function):
    """
    Context manager for temporarily using a different S3 client hook.
    """
    # Save current hook
    previous_hook = quilt3.hooks.get_build_s3_client_hook()
    
    try:
        # Install temporary hook
        quilt3.hooks.set_build_s3_client_hook(hook_function)
        yield
    finally:
        # Restore previous hook
        quilt3.hooks.set_build_s3_client_hook(previous_hook)

# Define a temporary hook
def debug_s3_client(*args, **kwargs):
    print("Creating S3 client with debug configuration")
    session = boto3.Session(*args, **kwargs)
    return session.client('s3', config=boto3.session.Config(log_level=logging.DEBUG))

# Use temporary hook
with temporary_s3_hook(debug_s3_client):
    bucket = quilt3.Bucket("s3://my-bucket")
    bucket.ls()  # Uses debug S3 client

# After context, original hook is restored
bucket2 = quilt3.Bucket("s3://another-bucket")  
bucket2.ls()  # Uses original S3 client configuration

Best Practices

Hook Implementation Guidelines

Maintain compatibility: Ensure custom hooks accept the same parameters as the default implementation
Handle errors gracefully: Include proper error handling in hook functions
Log appropriately: Add logging for debugging but avoid excessive output
Test thoroughly: Verify hook behavior in different scenarios
Document configuration: Document any environment variables or configuration requirements

Performance Considerations

Avoid heavy operations: Keep hook functions lightweight to prevent performance impact
Cache when appropriate: Consider caching expensive operations within hooks
Monitor resource usage: Be aware of memory and connection pool implications

Security Considerations

Validate inputs: Ensure hook functions validate any external inputs
Protect credentials: Avoid logging or exposing sensitive credential information
Use secure defaults: Default to secure configurations when possible

Error Handling

import quilt3.hooks
import boto3
import logging

def robust_s3_client_builder(*args, **kwargs):
    """
    Robust S3 client builder with comprehensive error handling.
    """
    try:
        session = boto3.Session(*args, **kwargs)
        s3_client = session.client('s3')
        
        # Test the client with a simple operation
        s3_client.list_buckets()
        
        return s3_client
        
    except Exception as e:
        logging.error(f"S3 client creation failed: {e}")
        
        # Fall back to default behavior or raise
        # Depending on your requirements:
        
        # Option 1: Re-raise the exception
        raise
        
        # Option 2: Return None to fall back to default
        # return None
        
        # Option 3: Try alternative configuration
        # return fallback_s3_client(*args, **kwargs)

# Install robust hook
quilt3.hooks.set_build_s3_client_hook(robust_s3_client_builder)

Install with Tessl CLI