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

Extension System

Name: tessl/pypi-cement
Author: tessl

The extension system provides framework extension loading and management capabilities. It enables modular functionality through well-defined extension points and automatic discovery, allowing applications to be enhanced with additional features.

Capabilities

Extension Handler Interface

Base interface for extension loading functionality that defines the contract for extension operations.

class ExtensionHandler:
    """
    Extension handler interface for loading and managing framework extensions.
    
    Provides methods for loading individual extensions and extension lists,
    enabling modular functionality enhancement.
    """
    
    def load_extension(self, ext_module: str) -> None:
        """
        Load a single extension module.
        
        Args:
            ext_module: Extension module name to load
        """
    
    def load_extensions(self, ext_list: List[str]) -> None:
        """
        Load multiple extension modules.
        
        Args:
            ext_list: List of extension module names to load
        """

Built-in Extensions

Cement provides many built-in extensions for common functionality:

argparse - Argument parsing and controller system
colorlog - Colorized console logging
configparser - Configuration file parsing
dummy - Basic/dummy handlers for all interfaces
generate - Code generation utilities
jinja2 - Jinja2 template engine integration
json - JSON output formatting
logging - Python logging integration
memcached - Memcached caching backend
mustache - Mustache template engine integration
print - Simple print-based output
redis - Redis caching and data backend
tabulate - Tabular output formatting
watchdog - File watching capabilities
yaml - YAML configuration and output support

Usage Examples

Basic Extension Loading

from cement import App, Controller, ex

class MyApp(App):
    class Meta:
        label = 'myapp'
        # Load extensions at startup
        extensions = [
            'colorlog',  # Colored logging
            'jinja2',    # Template engine
            'yaml',      # YAML support
            'redis'      # Redis caching
        ]

with MyApp() as app:
    app.setup()
    
    # Extensions are now available
    app.log.info('Application started with extensions')
    app.run()

Dynamic Extension Loading

from cement import App, Controller, ex

class ExtensionController(Controller):
    class Meta:
        label = 'ext'
        stacked_on = 'base'
        stacked_type = 'nested'
    
    @ex(
        help='load extension',
        arguments=[
            (['extension'], {'help': 'extension name to load'})
        ]
    )
    def load(self):
        """Load an extension dynamically."""
        extension_name = self.app.pargs.extension
        
        try:
            self.app.extend(extension_name, self.app)
            print(f'Successfully loaded extension: {extension_name}')
        except Exception as e:
            print(f'Failed to load extension {extension_name}: {e}')
    
    @ex(help='list loaded extensions')
    def list(self):
        """List currently loaded extensions."""
        if hasattr(self.app, '_loaded_extensions'):
            extensions = list(self.app._loaded_extensions.keys())
            print(f'Loaded extensions: {extensions}')
        else:
            print('No extensions loaded')

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

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

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

# Usage:
# myapp ext load colorlog
# myapp ext load jinja2
# myapp ext list

Conditional Extension Loading

from cement import App, Controller, ex, init_defaults
import os

CONFIG = init_defaults('myapp')
CONFIG['myapp']['environment'] = os.getenv('APP_ENV', 'development')
CONFIG['myapp']['enable_caching'] = True
CONFIG['myapp']['enable_templates'] = True

class ConditionalApp(App):
    class Meta:
        label = 'myapp'
        config_defaults = CONFIG
    
    def setup(self):
        # Load base extensions first
        base_extensions = ['colorlog', 'yaml']
        
        # Conditionally load extensions based on configuration
        extensions_to_load = base_extensions.copy()
        
        if self.config.get('myapp', 'enable_caching'):
            if os.getenv('REDIS_URL'):
                extensions_to_load.append('redis')
                print('Loading Redis extension (REDIS_URL found)')
            else:
                print('Caching enabled but no Redis URL found')
        
        if self.config.get('myapp', 'enable_templates'):
            extensions_to_load.append('jinja2')
            print('Loading Jinja2 extension (templates enabled)')
        
        env = self.config.get('myapp', 'environment')
        if env == 'development':
            extensions_to_load.append('watchdog')
            print('Loading Watchdog extension (development mode)')
        
        # Load all determined extensions
        for ext in extensions_to_load:
            try:
                self.extend(ext, self)
                print(f'✓ Loaded extension: {ext}')
            except Exception as e:
                print(f'✗ Failed to load extension {ext}: {e}')
        
        super().setup()

class BaseController(Controller):
    class Meta:
        label = 'base'
    
    @ex(help='show loaded extensions')
    def extensions(self):
        """Show information about loaded extensions."""
        print('Extension Status:')
        
        # Check commonly used extensions
        extensions_to_check = ['colorlog', 'yaml', 'redis', 'jinja2', 'watchdog']
        
        for ext in extensions_to_check:
            if hasattr(self.app, f'_{ext}_loaded'):
                print(f'  ✓ {ext}: Loaded')
            else:
                print(f'  ✗ {ext}: Not loaded')

with ConditionalApp() as app:
    app.run()

# Usage with environment variables:
# APP_ENV=production REDIS_URL=redis://localhost:6379 python myapp.py extensions

Custom Extension Creation

from cement import App, Handler, Interface
from cement.core.mail import MailHandler

class SlackInterface(Interface):
    """Interface for Slack messaging functionality."""
    
    class Meta:
        interface = 'slack'
    
    def send_message(self, channel, message):
        """Send message to Slack channel."""
        raise NotImplementedError

class SlackHandler(Handler, SlackInterface):
    """Handler for sending messages to Slack."""
    
    class Meta:
        interface = 'slack'
        label = 'slack'
        config_defaults = {
            'webhook_url': None,
            'username': 'MyApp Bot',
            'icon_emoji': ':robot_face:'
        }
    
    def send_message(self, channel, message):
        """Send message to Slack via webhook."""
        import requests
        import json
        
        webhook_url = self.app.config.get('slack', 'webhook_url')
        if not webhook_url:
            raise Exception('Slack webhook URL not configured')
        
        payload = {
            'channel': channel,
            'text': message,
            'username': self.app.config.get('slack', 'username'),
            'icon_emoji': self.app.config.get('slack', 'icon_emoji')
        }
        
        response = requests.post(webhook_url, data=json.dumps(payload))
        if response.status_code != 200:
            raise Exception(f'Slack API error: {response.status_code}')
        
        print(f'Message sent to #{channel}: {message}')

def load(app):
    """Extension load function."""
    # Define interface
    app.interface.define(SlackInterface)
    
    # Register handler
    app.handler.register(SlackHandler)

# Usage in application
from cement import Controller, ex, init_defaults

CONFIG = init_defaults('myapp', 'slack')
CONFIG['slack']['webhook_url'] = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'

class NotificationController(Controller):
    class Meta:
        label = 'notify'
        stacked_on = 'base'
        stacked_type = 'nested'
    
    @ex(
        help='send Slack notification',
        arguments=[
            (['channel'], {'help': 'Slack channel name'}),
            (['message'], {'help': 'message to send'})
        ]
    )
    def slack(self):
        """Send notification to Slack."""
        channel = self.app.pargs.channel
        message = self.app.pargs.message
        
        # Get Slack handler
        slack = self.app.handler.get('slack', 'slack')()
        slack.send_message(channel, message)

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

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

# Load custom extension
def load_custom_extensions(app):
    load(app)  # Load our custom Slack extension

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

# Usage:
# myapp notify slack general "Deployment completed successfully!"

Extension Configuration

from cement import App, init_defaults

CONFIG = init_defaults('myapp')

# Configure extensions
CONFIG['log.colorlog'] = {
    'level': 'DEBUG',
    'format': '%(log_color)s%(levelname)s%(reset)s - %(message)s',
    'colors': {
        'DEBUG': 'cyan',
        'INFO': 'green',
        'WARNING': 'yellow',
        'ERROR': 'red',
        'CRITICAL': 'bold_red'
    }
}

CONFIG['template.jinja2'] = {
    'template_dirs': ['./templates', './shared_templates'],
    'auto_reload': True,
    'cache_size': 50
}

CONFIG['cache.redis'] = {
    'host': 'localhost',
    'port': 6379,
    'db': 0,
    'default_timeout': 300
}

CONFIG['output.json'] = {
    'indent': 2,
    'sort_keys': True,
    'ensure_ascii': False
}

class ConfiguredApp(App):
    class Meta:
        label = 'myapp'
        config_defaults = CONFIG
        extensions = [
            'colorlog',
            'jinja2', 
            'redis',
            'json'
        ]
        
        # Use loaded extensions
        log_handler = 'colorlog'
        template_handler = 'jinja2'
        cache_handler = 'redis'
        output_handler = 'json'

with ConfiguredApp() as app:
    app.setup()
    
    # Extensions are configured and ready
    app.log.info('Application started with configured extensions')
    
    # Test cache
    app.cache.set('test_key', 'test_value', time=60)
    cached_value = app.cache.get('test_key')
    app.log.info(f'Cache test: {cached_value}')
    
    # Test output
    data = {'message': 'Hello from configured app', 'status': 'success'}
    output = app.render(data)
    print(output)
    
    app.run()

Extension Error Handling

from cement import App, Controller, ex

class RobustApp(App):
    class Meta:
        label = 'myapp'
    
    def setup(self):
        # Define required and optional extensions
        required_extensions = ['colorlog', 'yaml']
        optional_extensions = ['redis', 'jinja2', 'watchdog']
        
        # Load required extensions (fail if any fail)
        for ext in required_extensions:
            try:
                self.extend(ext, self)
                print(f'✓ Required extension loaded: {ext}')
            except Exception as e:
                print(f'✗ CRITICAL: Failed to load required extension {ext}: {e}')
                raise SystemExit(1)
        
        # Load optional extensions (continue if any fail)
        for ext in optional_extensions:
            try:
                self.extend(ext, self)
                print(f'✓ Optional extension loaded: {ext}')
            except Exception as e:
                print(f'⚠ Warning: Failed to load optional extension {ext}: {e}')
                print(f'  Continuing without {ext} functionality')
        
        super().setup()

class BaseController(Controller):
    class Meta:
        label = 'base'
    
    @ex(help='test extension availability')
    def test_extensions(self):
        """Test which extensions are available."""
        extensions_to_test = {
            'colorlog': 'Colored logging',
            'yaml': 'YAML configuration',
            'redis': 'Redis caching',
            'jinja2': 'Template rendering',
            'watchdog': 'File watching'
        }
        
        print('Extension Availability Test:')
        for ext, description in extensions_to_test.items():
            try:
                # Try to access extension functionality
                if ext == 'redis' and hasattr(self.app, 'cache'):
                    self.app.cache.set('test', 'value', time=1)
                    status = '✓ Available'
                elif ext == 'jinja2' and hasattr(self.app, 'template'):
                    status = '✓ Available'
                elif ext in ['colorlog', 'yaml']:  # Always loaded as required
                    status = '✓ Available'
                else:
                    status = '? Unknown'
            except Exception:
                status = '✗ Not available'
            
            print(f'  {ext:12} ({description:20}): {status}')

with RobustApp() as app:
    app.run()

# Usage:
# myapp test-extensions

Install with Tessl CLI