cjm-plugin-system 0.0.8

Generic plugin system with discovery, configuration validation, and runtime management for extensible Python applications.

cjm-plugin-system

Install

pip install cjm_plugin_system

Project Structure

nbs/
├── core/ (3)
│   ├── interface.ipynb  # Abstract base class defining the generic plugin interface
│   ├── manager.ipynb    # Plugin discovery, loading, and lifecycle management system
│   └── metadata.ipynb   # Data structures for plugin metadata
└── utils/ (1)
    └── validation.ipynb  # Validation helpers for plugin configuration dataclasses

Total: 4 notebooks across 2 directories

Module Dependencies

graph LR
    core_interface[core.interface<br/>Plugin Interface]
    core_manager[core.manager<br/>Plugin Manager]
    core_metadata[core.metadata<br/>Plugin Metadata]
    utils_validation[utils.validation<br/>Configuration Validation]

    core_interface --> utils_validation
    core_manager --> utils_validation
    core_manager --> core_metadata
    core_manager --> core_interface

4 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Plugin Interface (interface.ipynb)

Abstract base class defining the generic plugin interface

Import

from cjm_plugin_system.core.interface import (
    PluginInterface,
    PluginInterface_supports_streaming,
    PluginInterface_execute_stream
)

Functions

def PluginInterface_supports_streaming(self) -> bool: # True if execute_stream is implemented
    """Check if this plugin supports streaming execution."""
    # Default: check if execute_stream is overridden from the base class
    "Check if this plugin supports streaming execution."

def PluginInterface_execute_stream(
    self,
    *args, # Arguments for plugin execution
    **kwargs # Keyword arguments for plugin execution
) -> Generator[Any, None, Any]: # Yields partial results, returns final result
    "Stream execution results chunk by chunk."

Classes

class PluginInterface(ABC):
    "Generic plugin interface that all plugins must implement."
    
    def name(self) -> str: # Unique identifier for this plugin
            """Unique plugin identifier."""
            pass
    
        @property
        @abstractmethod
        def version(self) -> str: # Semantic version string (e.g., "1.0.0")
        "Unique plugin identifier."
    
    def version(self) -> str: # Semantic version string (e.g., "1.0.0")
            """Plugin version."""
            pass
    
        @abstractmethod
        def initialize(
            self,
            config:Optional[Any]=None # Configuration dataclass instance or dict
        ) -> None
        "Plugin version."
    
    def initialize(
            self,
            config:Optional[Any]=None # Configuration dataclass instance or dict
        ) -> None
        "Initialize the plugin with configuration."
    
    def execute(
            self,
            *args,
            **kwargs
        ) -> Any: # Plugin-specific output
        "Execute the plugin's main functionality."
    
    def is_available(self) -> bool: # True if all required dependencies are available
            """Check if the plugin's dependencies are available."""
            pass
    
        @abstractmethod
        def get_current_config(self) -> Any: # Current configuration dataclass instance
        "Check if the plugin's dependencies are available."
    
    def get_current_config(self) -> Any: # Current configuration dataclass instance
            """Return the current configuration state."""
            pass
    
        def get_config_defaults(self) -> Dict[str, Any]: # Default values from config_class
        "Return the current configuration state."
    
    def get_config_defaults(self) -> Dict[str, Any]: # Default values from config_class
            """Extract default values from the configuration dataclass."""
            if self.config_class is None
        "Extract default values from the configuration dataclass."
    
    def cleanup(self) -> None
        "Optional cleanup when plugin is unloaded."

Plugin Manager (manager.ipynb)

Plugin discovery, loading, and lifecycle management system

Import

from cjm_plugin_system.core.manager import (
    PluginManager,
    get_plugin_config_dataclass,
    get_all_plugin_config_dataclasses,
    get_plugin_config,
    get_plugin_config_class,
    validate_plugin_config,
    update_plugin_config,
    reload_plugin,
    execute_plugin_stream,
    check_streaming_support,
    get_streaming_plugins
)

Functions

def get_plugin_config_dataclass(
    self,
    plugin_name:str # Name of the plugin
) -> dataclass: # Current configuration dataclass
    "Get the configuration dataclass for a plugin."

def get_all_plugin_config_dataclasses(
    self
) -> Dict[str, dataclass]
    "Get configuration dataclasses for all loaded plugins."

def get_plugin_config(
    self,
    plugin_name:str # Name of the plugin
) -> Optional[Any]: # Current configuration dataclass instance or None if plugin not found
    "Get the current configuration of a plugin."

def get_plugin_config_class(
    self,
    plugin_name:str # Name of the plugin
) -> Optional[Type]: # Configuration dataclass type or None if plugin not found
    "Get the configuration dataclass type for a plugin."

def validate_plugin_config(
    self,
    plugin_name:str, # Name of the plugin
    config:Any # Configuration dataclass instance to validate
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Validate a configuration dataclass for a plugin."

def update_plugin_config(
    self,
    plugin_name:str, # Name of the plugin
    config:Any, # New configuration (dataclass instance or dict)
    merge:bool=True # Whether to merge with existing config or replace entirely
) -> bool: # True if successful, False otherwise
    "Update a plugin's configuration and reinitialize it."

def reload_plugin(
    self,
    plugin_name:str, # Name of the plugin to reload
    config:Optional[Any]=None # Optional new configuration (dataclass or dict)
) -> bool: # True if successful, False otherwise
    "Reload a plugin with optional new configuration."

def execute_plugin_stream(
    self,
    plugin_name:str, # Name of the plugin to execute
    *args, # Arguments to pass to the plugin
    **kwargs # Keyword arguments to pass to the plugin
) -> Generator[Any, None, Any]: # Generator yielding partial results, returns final result
    "Execute a plugin with streaming support if available."

def check_streaming_support(
    self,
    plugin_name:str # Name of the plugin to check
) -> bool: # True if plugin supports streaming
    "Check if a plugin supports streaming execution."

def get_streaming_plugins(
    self
) -> List[str]: # List of plugin names that support streaming
    "Get a list of all loaded plugins that support streaming."

Classes

class PluginManager:
    def __init__(
        self,
        plugin_interface:Type[PluginInterface]=PluginInterface, # Base class/interface plugins must implement
        entry_point_group:Optional[str]=None # Optional override for entry point group name
    )
    "Manages plugin discovery, loading, and lifecycle."
    
    def __init__(
            self,
            plugin_interface:Type[PluginInterface]=PluginInterface, # Base class/interface plugins must implement
            entry_point_group:Optional[str]=None # Optional override for entry point group name
        )
        "Initialize the plugin manager."
    
    def get_entry_points(self) -> importlib.metadata.EntryPoints: # Entry points for the configured group
            """Get plugin entry points from installed packages."""
            self.entry_points = []
            try
        "Get plugin entry points from installed packages."
    
    def discover_plugins(self) -> List[PluginMeta]: # List of discovered plugin metadata objects
            """Discover all installed plugins via entry points."""
            self.discovered = []
    
            for ep in self.entry_points
        "Discover all installed plugins via entry points."
    
    def load_plugin(
            self,
            plugin_meta:PluginMeta, # Plugin metadata
            config:Optional[Dict[str, Any]]=None # Optional configuration for the plugin
        ) -> bool: # True if successfully loaded, False otherwise
        "Load and initialize a plugin."
    
    def load_plugin_from_module(
            self,
            module_path:str, # Path to the Python module
            config:Optional[Dict[str, Any]]=None # Optional configuration for the plugin
        ) -> bool: # True if successfully loaded, False otherwise
        "Load a plugin directly from a Python module file or package."
    
    def unload_plugin(
            self,
            plugin_name:str # Name of the plugin to unload
        ) -> bool: # True if successfully unloaded, False otherwise
        "Unload a plugin and call its cleanup method."
    
    def get_plugin(
            self,
            plugin_name:str # Name of the plugin to retrieve
        ) -> Optional[PluginInterface]: # Plugin instance if found, None otherwise
        "Get a loaded plugin instance by name."
    
    def list_plugins(self) -> List[PluginMeta]: # List of metadata for all loaded plugins
            """List all loaded plugins."""
            return list(self.plugins.values())
    
        def execute_plugin(
            self,
            plugin_name:str, # Name of the plugin to execute
            *args, # Arguments to pass to the plugin
            **kwargs # Keyword arguments to pass to the plugin
        ) -> Any: # Result of the plugin execution
        "List all loaded plugins."
    
    def execute_plugin(
            self,
            plugin_name:str, # Name of the plugin to execute
            *args, # Arguments to pass to the plugin
            **kwargs # Keyword arguments to pass to the plugin
        ) -> Any: # Result of the plugin execution
        "Execute a plugin's main functionality."
    
    def enable_plugin(
            self,
            plugin_name:str # Name of the plugin to enable
        ) -> bool: # True if plugin was enabled, False if not found
        "Enable a plugin."
    
    def disable_plugin(
            self,
            plugin_name:str # Name of the plugin to disable
        ) -> bool: # True if plugin was disabled, False if not found
        "Disable a plugin without unloading it."

Plugin Metadata (metadata.ipynb)

Data structures for plugin metadata

Import

from cjm_plugin_system.core.metadata import (
    PluginMeta
)

Classes

@dataclass
class PluginMeta:
    "Metadata about a plugin."
    
    name: str  # Plugin's unique identifier
    version: str  # Plugin's version string
    description: str = ''  # Brief description of the plugin's functionality
    author: str = ''  # Plugin author's name or organization
    package_name: str = ''  # Python package name containing the plugin
    instance: Optional[Any]  # Plugin instance (PluginInterface subclass)
    enabled: bool = True  # Whether the plugin is enabled

Configuration Validation (validation.ipynb)

Validation helpers for plugin configuration dataclasses

Import

from cjm_plugin_system.utils.validation import (
    T,
    SCHEMA_TITLE,
    SCHEMA_DESC,
    SCHEMA_MIN,
    SCHEMA_MAX,
    SCHEMA_ENUM,
    SCHEMA_MIN_LEN,
    SCHEMA_MAX_LEN,
    SCHEMA_PATTERN,
    SCHEMA_FORMAT,
    validate_field_value,
    validate_config,
    config_to_dict,
    dict_to_config,
    extract_defaults
)

Functions

def validate_field_value(
    value:Any, # Value to validate
    metadata:Dict[str, Any], # Field metadata containing constraints
    field_name:str="" # Field name for error messages
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Validate a value against field metadata constraints."

def validate_config(
    config:Any # Configuration dataclass instance to validate
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Validate all fields in a configuration dataclass against their metadata constraints."

def config_to_dict(
    config:Any # Configuration dataclass instance
) -> Dict[str, Any]: # Dictionary representation of the configuration
    "Convert a configuration dataclass instance to a dictionary."

def dict_to_config(
    config_class:Type[T], # Configuration dataclass type
    data:Optional[Dict[str, Any]]=None, # Dictionary with configuration values
    validate:bool=False # Whether to validate against metadata constraints
) -> T: # Instance of the configuration dataclass
    "Create a configuration dataclass instance from a dictionary."

def extract_defaults(
    config_class:Type # Configuration dataclass type
) -> Dict[str, Any]: # Default values from the dataclass
    "Extract default values from a configuration dataclass type."

Variables

T
SCHEMA_TITLE = 'title'  # Display title for the field
SCHEMA_DESC = 'description'  # Help text description
SCHEMA_MIN = 'minimum'  # Minimum value for numbers
SCHEMA_MAX = 'maximum'  # Maximum value for numbers
SCHEMA_ENUM = 'enum'  # Allowed values for dropdowns
SCHEMA_MIN_LEN = 'minLength'  # Minimum string length
SCHEMA_MAX_LEN = 'maxLength'  # Maximum string length
SCHEMA_PATTERN = 'pattern'  # Regex pattern for strings
SCHEMA_FORMAT = 'format'  # String format (email, uri, date, etc.)