mm-pyplugin 2.3.2rc1

Python Plugin Project

mm-pyplugin

Plugin infrastructure for Python applications with Pydantic configuration validation, lifecycle management, and dependency injection.

Features

• Structured plugin lifecycle: __init__ → configure() → initialise() → teardown()
• Dependency injection: Context object for sharing services (loggers, databases, etc.) across plugins
• Pydantic validation: Type-safe configuration with automatic validation
• Fluent API: Method chaining for clean, readable plugin setup
• Hierarchical plugins: CompositePlugin base class for managing child plugins
• Builder pattern: Simplified construction with validation and initialization order enforcement
• Automatic discovery: Entry point-based plugin registration

Installation

pip install mm-pyplugin

For local development:

cd mm-pyplugin
uv sync

Quick Start

Simple Plugin

from mm_pyplugin import PluginBase
from pydantic import BaseModel
from typing import Type
import logging

class MyConfig(BaseModel):
    timeout: int = 30
    retries: int = 3

class MyPlugin(PluginBase):
    @classmethod
    def config_schema(cls) -> Type[BaseModel]:
        return MyConfig

    @property
    def config(self) -> MyConfig:
        return self._config_obj

    def initialise(self):
        # Access dependencies from context
        logger = self.context.get("logger")
        if logger:
            logger.info(f"Initializing with timeout={self.config.timeout}")

        self._initialised = True
        return self

    def teardown(self):
        logger = self.context.get("logger")
        if logger:
            logger.info("Cleaning up resources")

# Create dependency injection container
context = {
    "logger": logging.getLogger("app"),
    "database": Database("localhost:5432"),
    "event_bus": EventBus()
}

# Fluent API usage
plugin = (MyPlugin(context)
    .configure({"timeout": 60, "retries": 5})
    .initialise())

# Plugin can access shared services
plugin.context["logger"].info("Plugin ready")

# Later
plugin.teardown()

Dependency Injection (DI) Pattern

The context parameter is a dependency injection container that holds shared services:

# Create context with shared services
context = {
    "logger": logging.getLogger("app"),
    "database": Database("localhost:5432"),
    "cache": RedisCache("localhost:6379"),
    "metrics": MetricsCollector()
}

# All plugins share the same services
plugin1 = LogParser(context).configure(config1).initialise()
plugin2 = DataTransformer(context).configure(config2).initialise()
plugin3 = OutputWriter(context).configure(config3).initialise()

# All plugins use the same logger, database, etc.
plugin1.context["logger"].info("Parsing logs")
plugin2.context["database"].query("SELECT * FROM data")
plugin3.context["cache"].set("key", "value")

Benefits of DI:

• ✅ Share services across all plugins (single DB connection, etc.)
• ✅ Easy testing (inject mocks instead of real services)
• ✅ Flexible configuration (different contexts for dev/prod)

Hierarchical Plugin (CompositePlugin)

from mm_pyplugin import CompositePlugin

class DataTransformer(CompositePlugin):
    def __init__(self, context=None):
        super().__init__(context)
        # Create and register children (they inherit context automatically)
        self._parser = TimestampParser(context)
        self._mapper = LogLevelMapper(context)
        self._register_child(self._parser)
        self._register_child(self._mapper)

    @classmethod
    def config_schema(cls):
        return DataTransformerConfig

    @property
    def config(self):
        return self._config_obj

    def initialise(self):
        # Access shared logger
        logger = self.context.get("logger")
        if logger:
            logger.info("Initializing DataTransformer")

        # Initialize parent
        self._initialised = True
        # Then initialize all children
        self._initialise_children()
        return self

    def teardown(self):
        # Custom cleanup
        print("Parent cleanup")
        # Then teardown children in reverse order
        super().teardown()

# Usage - all plugins share same context
context = {"logger": logging.getLogger("app")}
transformer = DataTransformer(context)
transformer.configure({"output_format": "json"})
transformer._parser.configure(parser_config)
transformer._mapper.configure(mapper_config)
transformer.initialise()

# Cleanup (automatically handles children)
transformer.teardown()

Builder Pattern

from mm_pyplugin import PluginBuilder, CompositePluginBuilder

# Simple plugin
context = {"logger": logging.getLogger("app")}
plugin = (PluginBuilder(MyPlugin, context)
    .with_config({"timeout": 30})
    .build())

# Composite plugin with children
transformer = (CompositePluginBuilder(DataTransformer, context)
    .with_config({"output_format": "json"})
    .add_child(TimestampParser(context), {"timezone": "UTC"})
    .add_child(LogLevelMapper(context), {"case_sensitive": False})
    .build())

Architecture

Plugin Lifecycle

1. __init__(context=None) - Construct plugin with optional DI container
2. configure(config) - Load and validate configuration (returns self)
3. initialise() - Perform initialization, access self.context (returns self)
4. teardown() - Cleanup resources

Context (Dependency Injection)

The context is a dictionary that acts as a dependency injection container:

context = {
    "logger": logging.Logger,      # Shared logger
    "database": Database,           # Shared DB connection
    "event_bus": EventBus,          # Shared event bus
    "cache": Cache,                 # Shared cache
    "metrics": MetricsCollector,    # Shared metrics
    # ... any shared services
}

Plugins access context via self.context:

def initialise(self):
    logger = self.context.get("logger")
    db = self.context.get("database")

    if logger:
        logger.info("Plugin starting")
    if db:
        db.connect()

    self._initialised = True
    return self

Key Classes

• PluginBase - Abstract base class for all plugins

  • Enforces configuration schema via Pydantic
  • Provides fluent API for configuration and initialization
  • Stores and provides access to DI container via self.context
  • Requires teardown() implementation

• CompositePlugin - Base for hierarchical plugins

  • Manages child plugin lifecycle
  • Automatic teardown in reverse order
  • Children inherit parent's context automatically
  • Helper methods: _register_child(), _initialise_children()

• PluginBuilder - Builder for simple plugins

  • Validates configuration is set before build
  • Handles configuration and initialization in correct order

• CompositePluginBuilder - Builder for composite plugins

  • Manages child plugin registration
  • Ensures correct initialization order (parent → children)
  • Validates plugin class is CompositePlugin subclass

• PluginUtils - Plugin discovery and instantiation

  • find_plugins() - Discover plugins via entry points
  • create_plugin() - Instantiate, configure, and initialize plugins

Configuration

Plugins define their configuration schema using Pydantic models:

from pydantic import BaseModel

class MyPluginConfig(BaseModel):
    host: str = "localhost"
    port: int = 8080
    ssl_enabled: bool = False

Configuration can be loaded from:

• Dict: plugin.configure({"host": "example.com"})
• JSON file: plugin.configure("/path/to/config.json")
• Path object: plugin.configure(Path("config.json"))

Plugin Discovery

Register plugins via entry points in pyproject.toml:

[project.entry-points.mm_pyplugin]
my_plugin = "mypackage.plugins:MyPlugin"

Discover plugins:

from mm_pyplugin import PluginUtils

plugins = PluginUtils.find_plugins()
# {"mypackage.plugins.MyPlugin": <class MyPlugin>}

Testing

Running Tests

uv run python -m pytest

All 39 tests passing ✓

Testing Plugins with Mocks

# Production
prod_context = {
    "database": RealDatabase("prod-server"),
    "logger": RealLogger()
}
plugin = MyPlugin(prod_context).configure(config).initialise()

# Testing
test_context = {
    "database": MockDatabase(),  # Fake database
    "logger": MockLogger()       # Fake logger
}
plugin = MyPlugin(test_context).configure(config).initialise()

# Same plugin code, different dependencies!

Examples

See USAGE_EXAMPLES.md for comprehensive usage patterns and best practices.

Example plugins in src/mm_pyplugin/examples/:

• ExamplePlugin - Simple plugin demonstrating fluent API and DI
• ExampleParentPlugin - Composite plugin with child management and shared context

API Changes

New in this version:

• Context is now stored: Access via self.context throughout plugin lifecycle
• Simplified API: configure(config) and initialise() no longer require context parameter
• Context is optional: Can pass None or omit for plugins that don't need DI
• DI pattern: Context serves as dependency injection container for shared services

Migration from older versions:

# Old API
plugin = MyPlugin(ctx)
plugin.configure(ctx, config)
plugin.initialise(ctx)

# New API
plugin = MyPlugin(ctx)
plugin.configure(config)
plugin.initialise()

# Inside plugin - access context
def initialise(self):
    logger = self.context.get("logger")  # Access via self.context
    self._initialised = True
    return self

License

MIT