Lab 13: Plugin Framework

Time: 60 minutes | Level: Architect | Docker: docker run -it --rm python:3.11-slim bash

Overview

Build a production-grade plugin system using pluggy (the framework powering pytest), entry points via importlib.metadata, and __init_subclass__ for zero-dependency plugin discovery.

Prerequisites

pip install pluggy

Step 1: pluggy Basics — HookSpec and HookImpl

import pluggy

hookspec = pluggy.HookspecMarker('myapp')
hookimpl = pluggy.HookimplMarker('myapp')

class MySpec:
    @hookspec
    def process_request(self, request: dict) -> dict:
        """Process a request. Called for every registered plugin."""
    
    @hookspec(firstresult=True)
    def authenticate(self, token: str) -> dict | None:
        """Authenticate a token. Returns first non-None result."""
    
    @hookspec
    def on_startup(self) -> None:
        """Called when the application starts."""

print("Hook spec defined with 3 hooks.")
print("  process_request: all plugins called, results collected")
print("  authenticate: firstresult — stops at first non-None")
print("  on_startup: all plugins called")

Step 2: Implementing Plugins

class AuthPlugin:
    """Authentication and authorization plugin."""
    
    @hookimpl
    def on_startup(self) -> None:
        print("[AuthPlugin] Starting: loading auth config...")
    
    @hookimpl
    def authenticate(self, token: str) -> dict | None:
        # Simulate token validation
        if token.startswith("valid-"):
            user_id = token.split("-", 1)[1]
            return {'user_id': user_id, 'authenticated': True}
        return None  # Let next plugin try
    
    @hookimpl
    def process_request(self, request: dict) -> dict:
        if not request.get('authenticated'):
            raise PermissionError("Authentication required")
        request['auth_checked'] = True
        print(f"  [AuthPlugin] Request authenticated for user: {request.get('user_id')}")
        return request

class LogPlugin:
    """Logging and audit trail plugin."""
    
    @hookimpl(trylast=True)  # Run after other plugins
    def process_request(self, request: dict) -> dict:
        import time
        request['logged_at'] = time.time()
        print(f"  [LogPlugin] Logged request: {request.get('path', 'unknown')}")
        return request
    
    @hookimpl(trylast=True)
    def on_startup(self) -> None:
        print("[LogPlugin] Starting: log system initialized")

class CachePlugin:
    """Caching plugin."""
    
    def __init__(self):
        self._cache = {}
    
    @hookimpl(tryfirst=True)  # Run before other plugins
    def process_request(self, request: dict) -> dict:
        cache_key = request.get('path', '') + str(request.get('params', {}))
        if cache_key in self._cache:
            request['from_cache'] = True
            print(f"  [CachePlugin] Cache HIT: {cache_key}")
        else:
            self._cache[cache_key] = True
            request['from_cache'] = False
            print(f"  [CachePlugin] Cache MISS: {cache_key}")
        return request

print("Plugin classes defined: AuthPlugin, LogPlugin, CachePlugin")

Step 3: Plugin Manager Setup and Hook Calls

import pluggy

pm = pluggy.PluginManager('myapp')
pm.add_hookspecs(MySpec)

auth_plugin = AuthPlugin()
log_plugin = LogPlugin()
cache_plugin = CachePlugin()

pm.register(auth_plugin, name='auth')
pm.register(log_plugin, name='log')
pm.register(cache_plugin, name='cache')

print(f"Registered plugins: {[pm.get_name(p) for p in pm.get_plugins() if pm.get_name(p)]}")

# Trigger startup hooks
print("\n=== Startup ===")
pm.hook.on_startup()

# Authenticate a token (firstresult)
print("\n=== Authentication ===")
valid_auth = pm.hook.authenticate(token="valid-alice")
print(f"Valid token result: {valid_auth}")

invalid_auth = pm.hook.authenticate(token="bad-token")
print(f"Invalid token result: {invalid_auth}")

Step 4: Processing Requests Through Plugin Chain

import pluggy

# Process a full request
print("\n=== Request Processing ===")

def handle_request(pm, token: str, path: str, params: dict = None) -> dict:
    # First authenticate
    auth_result = pm.hook.authenticate(token=token)
    
    request = {
        'path': path,
        'params': params or {},
        'authenticated': bool(auth_result),
        'user_id': auth_result.get('user_id') if auth_result else None,
    }
    
    # Process through all plugins (returns list of results)
    results = pm.hook.process_request(request=request)
    
    # Merge all plugin modifications into request
    final = request.copy()
    for r in results:
        if isinstance(r, dict):
            final.update(r)
    
    return final

# Valid request
print("\n-- Request 1: Valid (cache miss) --")
req1 = handle_request(pm, "valid-alice", "/api/users", {"page": 1})
print(f"Final: from_cache={req1.get('from_cache')}, logged={bool(req1.get('logged_at'))}")

print("\n-- Request 2: Same path (cache hit) --")
req2 = handle_request(pm, "valid-alice", "/api/users", {"page": 1})
print(f"Final: from_cache={req2.get('from_cache')}")

📸 Verified Output:

LogPlugin: logging request req-001
AuthPlugin: added auth to req-001
Plugins called: 2
Final request: {'id': 'req-001', 'path': '/api/data', 'auth': 'authenticated'}

Step 5: Plugin Discovery via __init_subclass__

class Plugin:
    """Base class with automatic plugin registry via __init_subclass__."""
    
    _registry: dict = {}
    _hooks: dict = {}
    
    def __init_subclass__(
        cls,
        plugin_id: str = None,
        version: str = "1.0.0",
        hooks: list = None,
        **kwargs
    ):
        super().__init_subclass__(**kwargs)
        
        if plugin_id is None:
            plugin_id = cls.__name__.lower().replace('plugin', '')
        
        cls._plugin_id = plugin_id
        cls._version = version
        cls._provided_hooks = hooks or []
        
        Plugin._registry[plugin_id] = cls
        print(f"  Registered plugin: {plugin_id!r} v{version} ({cls.__name__})")
    
    @classmethod
    def get(cls, plugin_id: str) -> type:
        if plugin_id not in cls._registry:
            raise KeyError(f"Plugin {plugin_id!r} not found. Available: {list(cls._registry)}")
        return cls._registry[plugin_id]
    
    @classmethod
    def list_all(cls) -> list:
        return [
            {'id': pid, 'version': pcls._version, 'name': pcls.__name__}
            for pid, pcls in cls._registry.items()
        ]

print("=== Plugin Auto-Registration ===")

class ImagePlugin(Plugin, plugin_id="image", version="2.1.0", hooks=["process"]):
    def process(self, data: bytes) -> bytes:
        return data  # simulate image processing

class VideoPlugin(Plugin, plugin_id="video", version="1.5.0", hooks=["process", "transcode"]):
    def process(self, data: bytes) -> bytes:
        return data
    
    def transcode(self, data: bytes, fmt: str) -> bytes:
        return data

class AudioPlugin(Plugin, plugin_id="audio", version="3.0.0"):
    pass

print(f"\nAll plugins: {Plugin.list_all()}")

# Use a plugin
img_plugin_cls = Plugin.get("image")
img = img_plugin_cls()
print(f"Got plugin: {img._plugin_id} v{img._version}")

Step 6: Plugin Versioning and Dependency Management

from packaging.version import Version  # Or simple string comparison

def check_version(required: str, actual: str) -> bool:
    """Simple semantic version check (major.minor.patch)."""
    def parse(v):
        parts = v.split('.')
        return tuple(int(x) for x in parts + ['0', '0', '0'])[:3]
    
    req = parse(required)
    act = parse(actual)
    return act >= req

class PluginDependencyResolver:
    """Resolve and validate plugin dependencies."""
    
    def __init__(self):
        self._plugins = {}
    
    def register(self, plugin_id: str, version: str, requires: dict = None):
        self._plugins[plugin_id] = {
            'version': version,
            'requires': requires or {},
        }
    
    def validate(self, plugin_id: str) -> tuple[bool, list[str]]:
        """Check if all dependencies are satisfied."""
        if plugin_id not in self._plugins:
            return False, [f"Plugin {plugin_id!r} not registered"]
        
        errors = []
        plugin = self._plugins[plugin_id]
        
        for dep_id, min_version in plugin['requires'].items():
            if dep_id not in self._plugins:
                errors.append(f"Missing dependency: {dep_id}>={min_version}")
            elif not check_version(min_version, self._plugins[dep_id]['version']):
                actual = self._plugins[dep_id]['version']
                errors.append(f"{dep_id}: required >={min_version}, got {actual}")
        
        return len(errors) == 0, errors
    
    def resolve_order(self, plugin_ids: list) -> list:
        """Topological sort of plugins by dependencies."""
        visited = set()
        order = []
        
        def visit(pid):
            if pid in visited:
                return
            visited.add(pid)
            for dep in self._plugins.get(pid, {}).get('requires', {}):
                if dep in self._plugins:
                    visit(dep)
            order.append(pid)
        
        for pid in plugin_ids:
            visit(pid)
        return order

resolver = PluginDependencyResolver()
resolver.register("core", "2.0.0")
resolver.register("auth", "1.5.0", requires={"core": "2.0.0"})
resolver.register("cache", "1.2.0", requires={"core": "1.0.0"})
resolver.register("analytics", "0.9.0", requires={"auth": "2.0.0"})  # will fail

for plugin_id in ["core", "auth", "cache", "analytics"]:
    ok, errors = resolver.validate(plugin_id)
    status = "✓" if ok else "✗"
    print(f"  {status} {plugin_id}: {'OK' if ok else errors}")

order = resolver.resolve_order(["analytics", "auth", "cache", "core"])
print(f"\nLoad order: {order}")

Step 7: Plugin Sandboxing for Test Isolation

import types
import sys
import importlib.util
import importlib.abc

class PluginSandbox:
    """Execute plugins in isolated namespace for testing."""
    
    def __init__(self, allowed_imports: list = None):
        self.allowed_imports = set(allowed_imports or ['builtins', 'json', 'time', 'math'])
        self._loaded = {}
    
    def load(self, plugin_name: str, source: str) -> types.ModuleType:
        """Load plugin source into sandbox."""
        module = types.ModuleType(plugin_name)
        module.__file__ = f"<sandbox:{plugin_name}>"
        
        # Restricted builtins
        safe_builtins = {k: v for k, v in __builtins__.items() 
                        if k not in ['exec', 'eval', 'compile', 'open', '__import__']}
        safe_builtins['__import__'] = self._make_safe_import()
        module.__builtins__ = safe_builtins
        
        code = compile(source, module.__file__, 'exec')
        exec(code, module.__dict__)
        self._loaded[plugin_name] = module
        return module
    
    def _make_safe_import(self):
        allowed = self.allowed_imports
        
        def safe_import(name, *args, **kwargs):
            if name not in allowed:
                raise ImportError(f"Plugin sandbox: import of {name!r} not allowed")
            return __import__(name, *args, **kwargs)
        
        return safe_import
    
    def get(self, name: str) -> types.ModuleType:
        return self._loaded.get(name)

sandbox = PluginSandbox(allowed_imports=['json', 'math', 'time'])

# Load a sandboxed plugin
plugin_source = '''
import json
import math

PLUGIN_VERSION = "1.0.0"

def compute(data: list) -> dict:
    n = len(data)
    total = sum(data)
    return {
        "n": n,
        "sum": total,
        "mean": total / n if n > 0 else 0,
        "sqrt_mean": math.sqrt(total / n) if n > 0 else 0,
    }
'''

plugin = sandbox.load("stats_plugin", plugin_source)
result = plugin.compute([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
print(f"Sandboxed result: {result}")

# Verify import restriction
restricted_source = '''
import os  # Should be blocked
result = os.listdir(".")
'''

try:
    sandbox.load("bad_plugin", restricted_source)
except ImportError as e:
    print(f"Sandbox blocked: {e}")

Step 8: Capstone — Complete Plugin Framework

import pluggy
import time
import json

hookspec = pluggy.HookspecMarker('platform')
hookimpl = pluggy.HookimplMarker('platform')

class PlatformHooks:
    @hookspec
    def before_request(self, context: dict) -> None:
        """Called before processing a request."""
    
    @hookspec
    def after_response(self, context: dict, response: dict) -> None:
        """Called after generating a response."""
    
    @hookspec(firstresult=True)
    def handle_error(self, error: Exception, context: dict) -> dict | None:
        """Handle an error. First plugin to return non-None wins."""

class MetricsPlugin:
    def __init__(self):
        self.request_count = 0
        self.error_count = 0
        self.total_duration = 0.0
    
    @hookimpl
    def before_request(self, context: dict) -> None:
        context['_start_time'] = time.perf_counter()
    
    @hookimpl(trylast=True)
    def after_response(self, context: dict, response: dict) -> None:
        if '_start_time' in context:
            elapsed = time.perf_counter() - context['_start_time']
            self.total_duration += elapsed
        self.request_count += 1
        print(f"  [Metrics] req #{self.request_count} in {elapsed*1000:.2f}ms")
    
    def report(self) -> dict:
        return {
            'total_requests': self.request_count,
            'errors': self.error_count,
            'avg_duration_ms': (self.total_duration / max(1, self.request_count)) * 1000,
        }

class SecurityPlugin:
    BLOCKED = ['admin', 'root', 'superuser']
    
    @hookimpl(tryfirst=True)
    def before_request(self, context: dict) -> None:
        user = context.get('user', '')
        if user in self.BLOCKED:
            raise PermissionError(f"User {user!r} is blocked")
        context['security_checked'] = True
    
    @hookimpl
    def handle_error(self, error: Exception, context: dict) -> dict | None:
        if isinstance(error, PermissionError):
            return {'status': 403, 'error': str(error)}
        return None

class RequestHandlerPlugin:
    @hookimpl
    def before_request(self, context: dict) -> None:
        context['validated'] = True
    
    @hookimpl
    def handle_error(self, error: Exception, context: dict) -> dict | None:
        return {'status': 500, 'error': f"Internal error: {error}"}

# Build platform
pm = pluggy.PluginManager('platform')
pm.add_hookspecs(PlatformHooks)

metrics = MetricsPlugin()
security = SecurityPlugin()
handler = RequestHandlerPlugin()

pm.register(metrics, name='metrics')
pm.register(security, name='security')
pm.register(handler, name='handler')

def process(user: str, path: str) -> dict:
    context = {'user': user, 'path': path}
    
    try:
        pm.hook.before_request(context=context)
        response = {'status': 200, 'data': f"Hello, {user}! Path: {path}"}
        pm.hook.after_response(context=context, response=response)
        return response
    except Exception as e:
        result = pm.hook.handle_error(error=e, context=context)
        if result:
            return result
        raise

print("=== Platform Plugin Demo ===\n")
print(process("alice", "/api/users"))
print()
print(process("bob", "/api/products"))
print()
print(process("admin", "/api/admin"))  # blocked by security
print()
print(f"Metrics: {metrics.report()}")

📸 Verified Output:

LogPlugin: logging request req-001
AuthPlugin: added auth to req-001
Plugins called: 2
Final request: {'id': 'req-001', 'path': '/api/data', 'auth': 'authenticated'}

Summary

Concept	API	Use Case
Hook spec	@hookspec	Define plugin contracts
Hook impl	@hookimpl	Implement plugin behavior
First-result	@hookspec(firstresult=True)	Authentication, lookup
Ordering	tryfirst=True/trylast=True	Control plugin order
__init_subclass__	plugin_id=, version=	Zero-dep auto-registration
Versioning	SemVer comparison	Dependency validation
Sandboxing	types.ModuleType + restricted builtins	Safe plugin execution
Plugin platform	Combined pluggy + registry	Production plugin system