cjm-plugin-system

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

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  # JSON Schema validation helpers for plugin configuration

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 --> core_metadata
    core_manager --> core_interface

3 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[Dict[str, Any]]=None # Configuration dictionary for plugin-specific settings
        ) -> None
        "Plugin version."
    
    def initialize(
            self,
            config:Optional[Dict[str, Any]]=None # Configuration dictionary for plugin-specific settings
        ) -> 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
    
        @staticmethod
        @abstractmethod
        def get_config_schema() -> Dict[str, Any]: # JSON Schema describing configuration options
        "Check if the plugin's dependencies are available."
    
    def get_config_schema() -> Dict[str, Any]: # JSON Schema describing configuration options
            """Return JSON Schema describing the plugin's configuration options."""
            pass
    
        @abstractmethod
        def get_current_config(self) -> Dict[str, Any]: # Current configuration state
        "Return JSON Schema describing the plugin's configuration options."
    
    def get_current_config(self) -> Dict[str, Any]: # Current configuration state
            """Return the current configuration state."""
            pass
    
        def validate_config(
            self,
            config:Dict[str, Any] # Configuration to validate
        ) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
        "Return the current configuration state."
    
    def validate_config(
            self,
            config:Dict[str, Any] # Configuration to validate
        ) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
        "Validate a configuration dictionary against the schema."
    
    def get_config_defaults(self) -> Dict[str, Any]: # Default values from schema
            """Extract default values from the configuration schema."""
            schema = self.get_config_schema()
            return extract_defaults(schema)
    
        def cleanup(self) -> None
        "Extract default values from the configuration schema."
    
    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_schema,
    get_plugin_config,
    update_plugin_config,
    validate_plugin_config,
    get_all_plugin_schemas,
    reload_plugin,
    execute_plugin_stream,
    check_streaming_support,
    get_streaming_plugins
)

Functions

def get_plugin_config_schema(
    self,
    plugin_name:str # Name of the plugin
) -> Optional[Dict[str, Any]]: # Configuration schema or None if plugin not found
    "Get the configuration schema for a plugin."

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

def update_plugin_config(
    self,
    plugin_name:str, # Name of the plugin
    config:Dict[str, Any], # New configuration
    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 validate_plugin_config(
    self,
    plugin_name:str, # Name of the plugin
    config:Dict[str, Any] # Configuration to validate
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Validate a configuration dictionary for a plugin without applying it."

def get_all_plugin_schemas(
    self
) -> Dict[str, Dict[str, Any]]: # Dictionary mapping plugin names to their schemas
    "Get configuration schemas for all loaded plugins."

def reload_plugin(
    self,
    plugin_name:str, # Name of the plugin to reload
    config:Optional[Dict[str, Any]]=None # Optional new configuration
) -> 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`)

JSON Schema validation helpers for plugin configuration

Import

from cjm_plugin_system.utils.validation import (
    validate_config,
    extract_defaults
)

Functions

def validate_config(
    config:Dict[str, Any], # Configuration to validate
    schema:Dict[str, Any] # JSON Schema to validate against
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Validate a configuration dictionary against a JSON Schema."

def _basic_validate(
    config:Dict[str, Any], # Configuration to validate
    schema:Dict[str, Any] # JSON Schema to validate against
) -> Tuple[bool, Optional[str]]: # (is_valid, error_message)
    "Basic validation without jsonschema library."

def extract_defaults(
    schema:Dict[str, Any] # JSON Schema
) -> Dict[str, Any]: # Default values from schema
    "Extract default values from a JSON Schema."

Install

Project Structure

Module Dependencies

CLI Reference

Module Overview

Plugin Interface (interface.ipynb)

Import

Functions

Classes

Plugin Manager (manager.ipynb)

Import

Functions

Classes

Plugin Metadata (metadata.ipynb)

Import

Classes

Configuration Validation (validation.ipynb)

Import

Functions

Plugin Interface (`interface.ipynb`)

Plugin Manager (`manager.ipynb`)

Plugin Metadata (`metadata.ipynb`)

Configuration Validation (`validation.ipynb`)