tessl/pypi-tutor

The Docker-based Open edX distribution designed for peace of mind

Overview

Eval results

Files

Hooks and Plugin System

Name: tessl/pypi-tutor
Author: tessl

Event-driven plugin architecture enabling extensibility through Actions, Filters, and Contexts. The hooks system provides lifecycle management and data transformation capabilities, allowing plugins to customize every aspect of Tutor's behavior from configuration to deployment.

Capabilities

Action Hooks

Event callbacks that execute at specific lifecycle points without modifying data flow.

class Action[T]:
    """
    Action hook for event callbacks at specific lifecycle points.
    """
    def add(self, priority: int = None) -> Callable:
        """
        Decorator to add callback function with optional priority.
        
        Args:
            priority (int, optional): Execution priority (lower numbers run first)
            
        Returns:
            Callable: Decorator function
        """
    
    def do(self, *args, **kwargs) -> None:
        """
        Execute all registered callbacks with provided arguments.
        
        Args:
            *args: Positional arguments passed to callbacks
            **kwargs: Keyword arguments passed to callbacks
        """
    
    def do_from_context(self, context: Context, *args, **kwargs) -> None:
        """
        Execute callbacks from specific context only.
        
        Args:
            context (Context): Context to execute callbacks from
            *args: Positional arguments passed to callbacks
            **kwargs: Keyword arguments passed to callbacks
        """
    
    def clear(self, context: Context = None) -> None:
        """
        Clear callbacks from specific context or all contexts.
        
        Args:
            context (Context, optional): Context to clear, or None for all
        """

Filter Hooks

Data transformation chains that modify values as they pass through the system.

class Filter[T1, T2]:
    """
    Filter hook for data transformation chains.
    """
    def add(self, priority: int = None) -> Callable:
        """
        Decorator to add filter callback with optional priority.
        
        Args:
            priority (int, optional): Execution priority (lower numbers run first)
            
        Returns:
            Callable: Decorator function
        """
    
    def apply(self, value: T1, *args, **kwargs) -> T2:
        """
        Apply filter chain to transform the input value.
        
        Args:
            value (T1): Input value to transform
            *args: Additional arguments passed to filters
            **kwargs: Additional keyword arguments passed to filters
            
        Returns:
            T2: Transformed value after applying all filters
        """
    
    def add_item(self, item: Any) -> None:
        """
        Add single item to list-type filters.
        
        Args:
            item (Any): Item to add to the list
        """
    
    def add_items(self, *items: Any) -> None:
        """
        Add multiple items to list-type filters.
        
        Args:
            *items (Any): Items to add to the list
        """

Context System

Isolated execution contexts for organizing hooks by scope and lifecycle.

class Context:
    """
    Context for isolating hooks by scope and lifecycle.
    """
    def __init__(self, name: str):
        """
        Create a new context.
        
        Args:
            name (str): Context name identifier
        """

# Context instances
class Contexts:
    APP: Dict[str, Context]  # Per-application contexts
    PLUGINS: Context  # Plugin-specific context
    PLUGINS_V0_YAML: Context  # YAML v0 plugin context
    PLUGINS_V0_ENTRYPOINT: Context  # Entrypoint plugin context

Core Actions Catalog

Predefined actions for key lifecycle events in Tutor.

# Core lifecycle actions
COMPOSE_PROJECT_STARTED: Action[str, Config, str]  # Triggered when compose project starts
CONFIG_INTERACTIVE: Action[Config]  # After interactive configuration questions
CONFIG_LOADED: Action[Config]  # After configuration loading (read-only)
CORE_READY: Action[]  # Core system ready for plugin discovery
DO_JOB: Action[str, Any]  # Before job task execution
PLUGIN_LOADED: Action[str]  # Single plugin loaded
PLUGINS_LOADED: Action[]  # All plugins loaded
PLUGIN_UNLOADED: Action[str, str, Config]  # Plugin unloaded
PROJECT_ROOT_READY: Action[str]  # Project root directory ready

Core Filters Catalog

Predefined filters for data transformation throughout Tutor.

# Application and CLI filters
APP_PUBLIC_HOSTS: Filter[list[str], list[str]]  # Public application hostnames
CLI_COMMANDS: Filter[list[click.Command], list[click.Command]]  # CLI command list
CLI_DO_COMMANDS: Filter[list[Callable], list[Callable]]  # "do" subcommands
CLI_DO_INIT_TASKS: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Initialization tasks

# Configuration filters
CONFIG_DEFAULTS: Filter[list[tuple[str, Any]], list[tuple[str, Any]]]  # Default configuration
CONFIG_OVERRIDES: Filter[list[tuple[str, Any]], list[tuple[str, Any]]]  # Configuration overrides
CONFIG_UNIQUE: Filter[list[tuple[str, Any]], list[tuple[str, Any]]]  # Unique configuration values
CONFIG_USER: Filter[list[tuple[str, Any]], list[tuple[str, Any]]]  # User configuration

# Environment and template filters
ENV_PATCHES: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Template patches
ENV_PATTERNS_IGNORE: Filter[list[str], list[str]]  # Ignored template patterns
ENV_PATTERNS_INCLUDE: Filter[list[str], list[str]]  # Included template patterns
ENV_TEMPLATE_FILTERS: Filter[list[tuple[str, Callable]], list[tuple[str, Callable]]]  # Jinja2 filters
ENV_TEMPLATE_ROOTS: Filter[list[str], list[str]]  # Template root directories
ENV_TEMPLATE_TARGETS: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Template targets
ENV_TEMPLATE_VARIABLES: Filter[list[tuple[str, Any]], list[tuple[str, Any]]]  # Template variables

# Docker and image filters
DOCKER_BUILD_COMMAND: Filter[list[str], list[str]]  # Docker build command modification
IMAGES_BUILD: Filter[list[tuple], list[tuple]]  # Images to build
IMAGES_BUILD_REQUIRED: Filter[list[str], list[str]]  # Required images for build
IMAGES_BUILD_MOUNTS: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Build-time mounts
IMAGES_PULL: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Images to pull
IMAGES_PUSH: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Images to push

# Plugin and mount filters
COMPOSE_MOUNTS: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Compose bind mounts
MOUNTED_DIRECTORIES: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Auto-mounted directories
PLUGIN_INDEXES: Filter[list[str], list[str]]  # Plugin index URLs
PLUGINS_INFO: Filter[list[tuple[str, str]], list[tuple[str, str]]]  # Plugin information
PLUGINS_INSTALLED: Filter[list[str], list[str]]  # Installed plugins list
PLUGINS_LOADED: Filter[list[str], list[str]]  # Loaded plugins list

Plugin Management

Functions for installing, enabling, and managing plugins.

def is_installed(name: str) -> bool:
    """
    Check if a plugin is installed.
    
    Args:
        name (str): Plugin name
        
    Returns:
        bool: True if plugin is installed
    """

def iter_installed() -> Iterator[str]:
    """
    Iterate over installed plugin names.
    
    Returns:
        Iterator[str]: Iterator of installed plugin names
    """

def iter_info() -> Iterator[tuple[str, Optional[str]]]:
    """
    Iterate over plugin information (name, description).
    
    Returns:
        Iterator[tuple[str, Optional[str]]]: Iterator of (name, description) tuples
    """

def is_loaded(name: str) -> bool:
    """
    Check if a plugin is currently loaded.
    
    Args:
        name (str): Plugin name
        
    Returns:
        bool: True if plugin is loaded
    """

def load_all(names: Iterable[str]) -> None:
    """
    Load multiple plugins.
    
    Args:
        names (Iterable[str]): Plugin names to load
    """

def load(name: str) -> None:
    """
    Load a single plugin.
    
    Args:
        name (str): Plugin name to load
    """

def iter_loaded() -> Iterator[str]:
    """
    Iterate over loaded plugin names.
    
    Returns:
        Iterator[str]: Iterator of loaded plugin names
    """

def iter_patches(name: str) -> Iterator[str]:
    """
    Get template patches provided by a plugin.
    
    Args:
        name (str): Plugin name
        
    Returns:
        Iterator[str]: Iterator of patch names
    """

def unload(plugin: str) -> None:
    """
    Unload a plugin.
    
    Args:
        plugin (str): Plugin name to unload
    """

Hook Utilities

Utility functions for hook management and caching.

def clear_all(context: Context = None) -> None:
    """
    Clear all hooks from specific context or all contexts.
    
    Args:
        context (Context, optional): Context to clear, or None for all
    """

def lru_cache(func: Callable) -> Callable:
    """
    LRU cache decorator that clears when plugins change.
    
    Args:
        func (Callable): Function to cache
        
    Returns:
        Callable: Cached function
    """

# Priority constants
class priorities:
    HIGH = 10
    DEFAULT = 50  
    LOW = 90

Usage Examples

Creating Action Hooks

from tutor import hooks

# Define custom action
MY_CUSTOM_ACTION = hooks.Actions.create("my-custom-action")

# Add callback to action
@MY_CUSTOM_ACTION.add()
def my_callback(arg1: str, arg2: int) -> None:
    print(f"Action called with {arg1} and {arg2}")

# Trigger the action
MY_CUSTOM_ACTION.do("hello", 42)

Creating Filter Hooks

from tutor import hooks

# Define custom filter
MY_CUSTOM_FILTER = hooks.Filters.create("my-custom-filter")

# Add filter callback
@MY_CUSTOM_FILTER.add()
def my_filter(value: str, context: str) -> str:
    return f"{context}: {value.upper()}"

# Apply the filter
result = MY_CUSTOM_FILTER.apply("hello world", "debug")
# Result: "debug: HELLO WORLD"

Plugin Development

from tutor import hooks
from tutor.plugins import base

class MyPlugin(base.BasePlugin):
    def __init__(self):
        super().__init__("my-plugin", "My Custom Plugin")
    
    def configure(self) -> None:
        # Add configuration defaults
        hooks.Filters.CONFIG_DEFAULTS.add_items([
            ("MY_PLUGIN_SETTING", "default_value"),
            ("MY_PLUGIN_ENABLED", True),
        ])
        
        # Add template patches
        hooks.Filters.ENV_PATCHES.add_items([
            ("docker-compose.yml", "my-plugin/docker-compose.patch"),
            ("lms.yml", "my-plugin/lms.patch"),
        ])
        
        # Add CLI commands
        @hooks.Filters.CLI_COMMANDS.add()
        def add_my_command():
            return [self.create_command()]
        
        # React to lifecycle events
        @hooks.Actions.PLUGINS_LOADED.add()
        def on_plugins_loaded():
            print("All plugins loaded, initializing my plugin")

Working with Built-in Hooks

from tutor import hooks

# Add custom CLI command
@hooks.Filters.CLI_COMMANDS.add()  
def add_custom_command():
    import click
    
    @click.command()
    def my_command():
        """My custom command.""" 
        click.echo("Custom command executed!")
        
    return [my_command]

# Add configuration default
hooks.Filters.CONFIG_DEFAULTS.add_items([
    ("CUSTOM_SETTING", "my_value")
])

# React to configuration loading
@hooks.Actions.CONFIG_LOADED.add()
def on_config_loaded(config):
    print(f"Configuration loaded with {len(config)} settings")

Install with Tessl CLI