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  # 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

Loading

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."