tessl/pypi-cement

Advanced Application Framework for Python with a focus on Command Line Interfaces

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Hook System

Name: tessl/pypi-cement
Author: tessl

The hook system provides extension points throughout the application lifecycle, enabling custom functionality injection at defined points in application execution. Hooks allow developers to extend and customize application behavior without modifying core framework code.

Capabilities

Hook Manager

Manages application hooks including hook definition, registration, and execution.

class HookManager:
    """
    Manages application hooks for extending functionality at defined points.
    
    The hook manager provides centralized control over hook registration,
    execution, and lifecycle management.
    """
    
    def define(self, name: str) -> None:
        """
        Define a new hook in the system.
        
        Args:
            name: Hook name to define
        """
    
    def defined(self, name: str) -> bool:
        """
        Check if a hook is defined.
        
        Args:
            name: Hook name to check
            
        Returns:
            True if hook is defined, False otherwise
        """
    
    def register(self, name: str, func: Callable, weight: int = 0) -> None:
        """
        Register a function to run when hook is executed.
        
        Args:
            name: Hook name to register for
            func: Function to execute when hook runs
            weight: Execution weight (lower weights run first)
        """
    
    def run(self, name: str, app: 'App', *args: Any, **kwargs: Any) -> None:
        """
        Execute all registered functions for a hook.
        
        Args:
            name: Hook name to execute
            app: Application instance
            *args: Additional arguments passed to hook functions
            **kwargs: Additional keyword arguments passed to hook functions
        """

Built-in Hook Points

The framework provides several built-in hooks for extending functionality:

pre_setup - Before application setup
post_setup - After application setup
pre_run - Before application run
post_run - After application run
pre_close - Before application close
post_close - After application close
signal - When signals are caught
pre_render - Before output rendering
post_render - After output rendering

Usage Examples

Registering Hook Functions

from cement import App, Controller, ex

def my_pre_setup_hook(app):
    """Hook function that runs before app setup."""
    print('Running pre-setup initialization')
    # Initialize custom components
    app.custom_data = {'initialized': True}

def my_post_setup_hook(app):
    """Hook function that runs after app setup."""
    print('Running post-setup configuration')
    # Validate configuration
    if not hasattr(app, 'custom_data'):
        raise Exception('Custom data not initialized')

def my_pre_run_hook(app):
    """Hook function that runs before app execution."""
    print(f'Starting application: {app._meta.label}')
    # Log application start
    app.log.info('Application execution started')

def my_post_run_hook(app):
    """Hook function that runs after app execution."""
    print('Application execution completed')
    # Cleanup or final logging
    app.log.info('Application execution completed')

class BaseController(Controller):
    class Meta:
        label = 'base'
    
    @ex(help='test command')
    def test(self):
        """Test command to demonstrate hooks."""
        print('Executing test command')
        print(f'Custom data: {self.app.custom_data}')

class MyApp(App):
    class Meta:
        label = 'myapp'
        base_controller = 'base'
        handlers = [BaseController]

def load_hooks(app):
    """Load application hooks."""
    app.hook.register('pre_setup', my_pre_setup_hook)
    app.hook.register('post_setup', my_post_setup_hook)
    app.hook.register('pre_run', my_pre_run_hook)
    app.hook.register('post_run', my_post_run_hook)

with MyApp() as app:
    load_hooks(app)
    app.run()

Hook Weights and Execution Order

from cement import App

def first_hook(app):
    """Hook with lowest weight runs first."""
    print('First hook (weight: -10)')

def middle_hook(app):
    """Hook with default weight."""
    print('Middle hook (weight: 0)')

def last_hook(app):
    """Hook with highest weight runs last."""
    print('Last hook (weight: 10)')

class MyApp(App):
    class Meta:
        label = 'myapp'

def load_hooks(app):
    """Load hooks with different weights."""
    app.hook.register('pre_setup', first_hook, weight=-10)
    app.hook.register('pre_setup', middle_hook, weight=0)  
    app.hook.register('pre_setup', last_hook, weight=10)

with MyApp() as app:
    load_hooks(app)
    app.setup()
    app.run()

# Output:
# First hook (weight: -10)
# Middle hook (weight: 0)
# Last hook (weight: 10)

Custom Hook Definition

from cement import App, Controller, ex

def database_connect_hook(app, connection_string):
    """Custom hook for database connection."""
    print(f'Connecting to database: {connection_string}')
    # Simulate database connection
    app.db_connected = True

def database_migrate_hook(app):
    """Custom hook for database migration."""
    if not getattr(app, 'db_connected', False):
        raise Exception('Database not connected')
    print('Running database migrations')
    app.db_migrated = True

def validate_database_hook(app):
    """Custom hook for database validation."""
    if not getattr(app, 'db_migrated', False):
        raise Exception('Database not migrated')
    print('Database validation successful')

class DatabaseController(Controller):
    class Meta:
        label = 'database'
        stacked_on = 'base'
        stacked_type = 'nested'
    
    @ex(help='initialize database')
    def init(self):
        """Initialize database system."""
        # Define and run custom hooks
        if not self.app.hook.defined('database_connect'):
            self.app.hook.define('database_connect')
        if not self.app.hook.defined('database_migrate'):
            self.app.hook.define('database_migrate')
        if not self.app.hook.defined('database_validate'):
            self.app.hook.define('database_validate')
        
        # Run hooks in sequence
        self.app.hook.run('database_connect', self.app, 'postgresql://localhost/myapp')
        self.app.hook.run('database_migrate', self.app)
        self.app.hook.run('database_validate', self.app)
        
        print('Database initialization completed')

class BaseController(Controller):
    class Meta:
        label = 'base'

class MyApp(App):
    class Meta:
        label = 'myapp'
        base_controller = 'base'
        handlers = [BaseController, DatabaseController]

def load_hooks(app):
    """Load custom database hooks."""
    # Register hooks (they will be defined when needed)
    app.hook.register('database_connect', database_connect_hook)
    app.hook.register('database_migrate', database_migrate_hook)
    app.hook.register('database_validate', validate_database_hook)

with MyApp() as app:
    load_hooks(app)
    app.run()

# Usage:
# myapp database init

Extension Loading Hooks

from cement import App

def load_redis_extension(app):
    """Hook to load Redis extension if needed."""
    if app.config.get('myapp', 'cache_handler') == 'redis':
        print('Loading Redis extension for caching')
        app.extend('redis', app)

def load_jinja2_extension(app):
    """Hook to load Jinja2 extension if needed."""
    if app.config.get('myapp', 'template_handler') == 'jinja2':
        print('Loading Jinja2 extension for templating')
        app.extend('jinja2', app)

def validate_extensions(app):
    """Hook to validate required extensions are loaded."""
    required_extensions = ['colorlog', 'yaml']
    for ext in required_extensions:
        if ext not in app.loaded_extensions:
            print(f'Warning: Required extension {ext} not loaded')

class MyApp(App):
    class Meta:
        label = 'myapp'
        extensions = ['colorlog', 'yaml']

def load_hooks(app):
    """Load extension management hooks."""
    app.hook.register('post_setup', load_redis_extension, weight=-5)
    app.hook.register('post_setup', load_jinja2_extension, weight=-5)
    app.hook.register('post_setup', validate_extensions, weight=5)

with MyApp() as app:
    load_hooks(app)
    app.setup()
    app.run()

Configuration Validation Hooks

from cement import App, init_defaults

CONFIG = init_defaults('myapp', 'database')
CONFIG['myapp']['environment'] = 'development'
CONFIG['database']['host'] = 'localhost'

def validate_environment_config(app):
    """Validate environment-specific configuration."""
    env = app.config.get('myapp', 'environment')
    
    if env == 'production':
        required_keys = ['database.host', 'database.password', 'database.ssl']
        for key_path in required_keys:
            section, key = key_path.split('.')
            try:
                value = app.config.get(section, key)
                if not value:
                    raise ValueError(f'Production environment requires {key_path}')
            except KeyError:
                raise ValueError(f'Production environment requires {key_path}')
        print('Production configuration validated')
    else:
        print(f'Configuration validated for {env} environment')

def setup_logging_config(app):
    """Configure logging based on environment."""
    env = app.config.get('myapp', 'environment')
    
    if env == 'production':
        app.log.set_level('WARNING')
        print('Logging set to WARNING level for production')
    else:
        app.log.set_level('DEBUG')
        print('Logging set to DEBUG level for development')

class MyApp(App):
    class Meta:
        label = 'myapp'
        config_defaults = CONFIG

def load_hooks(app):
    """Load configuration hooks."""
    app.hook.register('post_setup', validate_environment_config, weight=-10)
    app.hook.register('post_setup', setup_logging_config, weight=-5)

with MyApp() as app:
    load_hooks(app)
    app.setup()
    app.run()

Signal Handling Hooks

from cement import App
import signal

def cleanup_resources(app, signum, frame):
    """Hook to cleanup resources when signal received."""
    print(f'Received signal {signum}, cleaning up resources')
    
    # Close database connections
    if hasattr(app, 'db_connection'):
        app.db_connection.close()
        print('Database connection closed')
    
    # Save application state
    if hasattr(app, 'state'):
        with open('/tmp/app_state.json', 'w') as f:
            json.dump(app.state, f)
        print('Application state saved')

def log_signal_received(app, signum, frame):
    """Hook to log signal reception."""
    signal_names = {
        signal.SIGTERM: 'SIGTERM',
        signal.SIGINT: 'SIGINT',
        signal.SIGHUP: 'SIGHUP'
    }
    signal_name = signal_names.get(signum, f'Signal {signum}')
    app.log.warning(f'Received {signal_name}, initiating shutdown')

class MyApp(App):
    class Meta:
        label = 'myapp'
        catch_signals = [signal.SIGTERM, signal.SIGINT, signal.SIGHUP]

def load_hooks(app):
    """Load signal handling hooks."""
    app.hook.register('signal', cleanup_resources, weight=-10)
    app.hook.register('signal', log_signal_received, weight=-5)

with MyApp() as app:
    load_hooks(app)
    app.setup()
    
    # Simulate some application state
    app.state = {'last_action': 'startup', 'user_count': 0}
    
    app.run()

Render Hooks for Output Processing

from cement import App, Controller, ex
import json

def add_metadata_to_output(app, data, rendered_output):
    """Hook to add metadata to rendered output."""
    if isinstance(data, dict):
        # Add timestamp and version metadata
        import datetime
        data['_metadata'] = {
            'timestamp': datetime.datetime.now().isoformat(),
            'version': app.config.get('myapp', 'version', '1.0.0'),
            'environment': app.config.get('myapp', 'environment', 'development')
        }

def log_render_operation(app, data, rendered_output):
    """Hook to log render operations."""
    data_size = len(str(data))
    output_size = len(rendered_output)
    app.log.debug(f'Rendered {data_size} bytes of data to {output_size} bytes of output')

class BaseController(Controller):
    class Meta:
        label = 'base'
    
    @ex(help='get user information')
    def user_info(self):
        """Get user information with metadata."""
        user_data = {
            'id': 123,
            'name': 'John Doe',
            'email': 'john@example.com',
            'status': 'active'
        }
        
        # Render with hooks
        output = self.app.render(user_data)
        print(output)

class MyApp(App):
    class Meta:
        label = 'myapp'
        base_controller = 'base'
        handlers = [BaseController]

def load_hooks(app):
    """Load render hooks."""
    app.hook.register('pre_render', add_metadata_to_output)
    app.hook.register('post_render', log_render_operation)

with MyApp() as app:
    load_hooks(app)
    app.run()

# Usage:
# myapp user-info

Install with Tessl CLI