paulpas Framework Extension Design
Acts as a senior framework architect designing extensibility surfaces for frameworks that YOUR team builds and maintains. When loaded, the model defines stable plugin contracts, middleware pipeline architectures, versioned public APIs, and extension authoring guides — ensuring that third-party contributors can extend the framework without breaking changes, without accessing internal state, and with clear upgrade paths. This is NOT about using extensions in an existing framework like Django or Rails; this is about BUILDING a framework whose primary purpose is to be extended by others.
TL;DR Checklist
- Every public extension interface has explicit input/output contracts with typed signatures
- Extension lifecycle phases (install, activate, execute, teardown) are clearly defined and tested
- Middleware pipeline uses composable
(context, next) -> resultsignature with priority ordering - Extension API follows semantic versioning — breaking changes require major version bump
- Documentation includes a complete "Hello World" plugin that third parties can copy-paste-run
- Backward compatibility contract is tested via snapshot tests and deprecation warnings
When to Use
Use this skill when:
- Building an internal framework or SDK that your organization's teams will extend with custom plugins
- Designing a public-facing plugin marketplace where third-party developers need stable, well-documented extension points
- Creating a middleware pipeline system where users inject custom handlers at specific lifecycle phases
- Architecting a CLI framework (like
uvicorn,pytest, orinvoke) that external tools can extend - Refactoring an existing framework to expose clean extension surfaces instead of requiring monkey-patching
When NOT to Use
Avoid this skill for:
- Using extensions in your own application — if you're consuming plugins from a framework like Django, use
framework-driven-designinstead - Integrating external frameworks into your codebase — use
framework-integration-patternswhen adapting third-party libraries - Designing the overall project structure — that belongs in
framework-architecture-design, which handles directory layout and module boundaries - Creating simple callback mechanisms — if you only need one function to run after an event, a plain Python callable registration is simpler than a full plugin system
- API versioning for REST endpoints — use API design patterns for client-facing HTTP APIs; extension design is about framework-level hooks, not resource URLs
Core Workflow
Catalog Extension Surface Areas — Identify every place in the framework where external code needs to hook in. Categorize each surface into one of four types: (a) Lifecycle Hooks — events fired at specific framework phases (boot, request-start, request-end, shutdown); (b) Middleware/Interceptors — composable handlers that process requests or data as they pass through the pipeline; (c) Provider/Service Extension — custom implementations of framework interfaces (repositories, formatters, validators); (d) Configuration Extension — adding new configuration schemas and defaults. For each surface, define what information is available to the extension author and what they can modify or return.
Checkpoint: For every extension surface, write a one-sentence "what this lets plugin authors do" description. If you cannot articulate it without mentioning internal framework details, the surface is leaking implementation concerns.
Design Plugin Interface Contracts — For each extension type, define the exact interface (Protocol, ABC, or language-native equivalent) that plugins must implement. Every method must have: typed signatures with explicit parameter types, docstrings describing purpose and expected behavior, return type contracts, and documented exception types that may be raised. The interface is the framework's promise to plugin authors — once published in a stable version, it cannot change without a major version bump. Design these interfaces following the 5 Laws of Elegant Defense (from
code-philosophy): ensure data flows naturally through extension boundaries (data flow), each layer owns its state (early exit and fail-fast on invalid inputs), and keep interfaces focused so they guide the developer toward correct usage (intentional naming).Checkpoint: Write three independent plugin implementations from scratch using only the interface definition. If any implementation requires reading internal framework source code, the interface contract is incomplete.
Design Middleware Pipeline Architecture — Create a composable pipeline where middleware components chain via
(context, next) -> result. Each middleware receives a context object (immutable during its execution), can modify it before callingnext(), and can observe or mutate the response afternext()returns. Support priority ordering, conditional execution, and early termination (short-circuit). The pipeline itself is immutable at runtime — plugins register during framework bootstrap.Checkpoint: Verify that no middleware can bypass another by observing whether the context object is shared but not mutable between stages. Test with three middleware pieces: one that short-circuits, one that runs after it (should never execute), and one that runs before it (should always execute).
Version Extension APIs with Semantic Versioning — Establish a clear versioning policy for each extension surface. Minor versions may add new optional methods to interfaces; major versions may change method signatures or remove deprecated methods. Provide deprecation warnings (runtime logging + type-level annotations) at least one major version before removal. Document the exact compatibility matrix: "Plugin built against v2.x works with framework v2.x and v3.x; plugin built against v1.x requires migration to v2."
Checkpoint: Write a migration guide snippet showing how a plugin author updates from interface version N to version N+1, including what code changes are required and what is automatically compatible.
Create Extension Authoring Guide — Document the complete developer experience for building plugins: installation, registration, testing, debugging, publishing. Include a minimum viable "Hello World" plugin that works out of the box. Document common pitfalls (shared mutable state, blocking calls in async pipelines, missing cleanup) and how to avoid them.
Checkpoint: Give only this guide to an unfamiliar developer. They should be able to build, register, and test a working plugin within one hour without reading the framework's internal source code.
Implementation Patterns
Pattern 1: Plugin Interface Contract with Lifecycle Management
Define stable plugin interfaces that plugins implement, and a manager that controls their lifecycle from registration through execution to teardown. This pattern works across all languages — shown in Python but the concepts apply identically to TypeScript (interfaces), Go (interfaces), Java (interfaces/abstract classes), and Rust (traits).
# framework/plugins/contract.py — The PUBLIC API surface. Plugin authors import ONLY from here.
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Protocol
class PluginState(Enum):
"""Lifecycle states for a plugin instance."""
INSTALLED = "installed" # Plugin is registered but not yet activated
ACTIVE = "active" # Plugin has been initialized and is running
SUSPENDED = "suspended" # Plugin is paused (can be resumed)
ERROR = "error" # Plugin encountered a fatal error
class PluginMetadata(Protocol):
"""Metadata that every plugin must report about itself."""
@property
def name(self) -> str: ...
@property
def version(self) -> str: ...
@property
def description(self) -> str: ...
@property
def state(self) -> PluginState: ...
class LifecycleManager(Protocol):
"""Methods available to plugins during their lifecycle."""
def register_shutdown_hook(self, callback: Any) -> None: ...
def get_config(self, key: str) -> Any: ...
def emit_event(self, event_name: str, payload: dict[str, Any]) -> None: ...
class Plugin(ABC):
"""Base class for all framework plugins.
Plugin authors MUST subclass this and implement the required methods.
The framework calls lifecycle hooks in strict order:
on_install() → on_activate() → [runtime] → on_deactivate() → on_uninstall()
All methods receive a LifecycleManager to interact with the framework
without accessing internal state directly.
"""
# Override these in subclasses
name: str = "unnamed-plugin"
version: str = "0.0.0"
description: str = ""
@property
def state(self) -> PluginState:
"""Current lifecycle state (managed by the framework, not the plugin)."""
return self._state # type: ignore[attr-defined]
def __init__(self) -> None:
self._state: PluginState = PluginState.INSTALLED
# --- Lifecycle Hooks (called by framework in order) ---
def on_install(self, manager: LifecycleManager) -> None:
"""Called when plugin is first registered.
Use this to validate configuration, create database tables, or
perform one-time setup. Raise RuntimeError for fatal errors that
should prevent the plugin from being installed.
"""
pass # Optional override — no-op by default
def on_activate(self, manager: LifecycleManager) -> None:
"""Called when the framework is starting up and plugins become active.
Use this to register event listeners, start background threads, or
initialize connections. Plugins should NOT perform heavy computation here —
defer work until first use.
"""
self._state = PluginState.ACTIVE
def on_deactivate(self, manager: LifecycleManager) -> None:
"""Called when the framework is shutting down.
Use this to close connections, flush buffers, and release resources.
Must be idempotent — called once per activation cycle, but plugins
may be deactivated/activated multiple times.
"""
self._state = PluginState.INSTALLED
def on_uninstall(self, manager: LifecycleManager) -> None:
"""Called when the plugin is being removed entirely.
Clean up everything created in on_install. After this call,
no references to this plugin instance should remain.
"""
self._state = PluginState.INSTALLED
# --- Runtime Hooks (called during normal operation) ---
def handle_event(self, manager: LifecycleManager, event_name: str, payload: dict[str, Any]) -> Any:
"""Called when a framework event is emitted that this plugin cares about.
Return value is passed to the next handler in the chain (if applicable).
Raise an exception to halt the event chain and mark the plugin ERROR.
"""
return payload
# --- Concrete Example Plugin (from plugin author) ---
class AuditLogPlugin(Plugin):
"""Records all framework events to a structured audit log."""
name = "audit-log"
version = "1.2.0"
description = "Structured event auditing for compliance and debugging"
def __init__(self, log_backend: str = "stdout") -> None:
super().__init__()
self._log_backend = log_backend
self._events: list[dict[str, Any]] = []
def on_activate(self, manager: LifecycleManager) -> None:
super().on_activate(manager)
# Register for specific events the framework emits
self._events_config = {
"request.started": True,
"request.completed": True,
"error.occurred": True,
}
def handle_event(self, manager: LifecycleManager, event_name: str, payload: dict[str, Any]) -> Any:
if not self._events_config.get(event_name):
return payload # Ignore irrelevant events
audit_entry = {
"plugin": self.name,
"event": event_name,
"payload": payload,
"timestamp": manager.get_config("audit.timestamp"), # type: ignore[union-attr]
}
self._events.append(audit_entry)
if self._log_backend == "stdout":
print(f"[AUDIT] {event_name}: {payload}")
return payload
# --- Plugin Discovery and Registration (framework internal, NOT part of public API) ---
class PluginRegistry:
"""Internal framework component that manages plugin lifecycle.
This class is NOT importable by plugin authors. It uses the Plugin
protocol to interact with plugins without exposing framework internals.
"""
def __init__(self) -> None:
self._plugins: dict[str, Plugin] = {}
def register(self, plugin: Plugin) -> None:
if plugin.name in self._plugins:
raise ValueError(f"Plugin '{plugin.name}' already registered")
self._plugins[plugin.name] = plugin
plugin.on_install(FrameworkLifecycleBridge()) # type: ignore[name-defined]
def activate_all(self) -> None:
for plugin in self._plugins.values():
if plugin.state == PluginState.INSTALLED:
plugin.on_activate(FrameworkLifecycleBridge()) # type: ignore[name-defined]
def deactivate_all(self) -> None:
for name, plugin in list(self._plugins.items()):
if plugin.state == PluginState.ACTIVE:
plugin.on_deactivate(FrameworkLifecycleBridge()) # type: ignore[name-defined]
def shutdown(self) -> None:
self.deactivate_all()
for name, plugin in list(self._plugins.items()):
plugin.on_uninstall(FrameworkLifecycleBridge()) # type: ignore[name-defined]
del self._plugins[name]
class FrameworkLifecycleBridge(LifecycleManager):
"""Bridge between framework internals and the public LifecycleManager interface.
Plugin authors NEVER see this class — it's injected automatically by the registry.
It translates calls from the clean protocol into framework-specific operations.
"""
def register_shutdown_hook(self, callback: Any) -> None:
import atexit
atexit.register(callback)
def get_config(self, key: str) -> Any:
# Access to a controlled configuration subset only
return {"audit.timestamp": "2024-01-01T00:00:00Z"}.get(key) # type: ignore[return-value]
def emit_event(self, event_name: str, payload: dict[str, Any]) -> None:
# Internal framework event emission — plugin authors cannot call this directly
pass # Implementation details hidden from plugin API
Pattern 2: Middleware Pipeline with Priority and Short-Circuit Support
Middleware pipelines are the most common extension surface in modern frameworks. This pattern provides a composable, type-safe pipeline where middleware can process requests before and after downstream handlers, support priority ordering, and short-circuit when appropriate.
# framework/middleware/pipeline.py — Public API for middleware pipeline design
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
@dataclass(frozen=True)
class MiddlewareContext:
"""Immutable context object passed through the middleware chain.
Middleware MAY attach metadata to this context using the mutable wrapper.
The core fields (request, method, path, headers) are immutable — they
represent the original incoming request and cannot be spoofed by plugins.
"""
method: str = "GET"
path: str = "/"
headers: dict[str, str] = field(default_factory=dict)
body: bytes | None = None
_metadata: dict[str, Any] = field(default_factory=dict, repr=False)
def get(self, key: str, default: Any = None) -> Any:
"""Read metadata attached by previous middleware in the chain."""
return self._metadata.get(key, default)
def set(self, key: str, value: Any) -> None:
"""Attach metadata for downstream middleware to read.
This is the ONLY way middleware communicate with each other through
the pipeline. Never store data on 'self' — it may be shared across
requests in connection-pooled environments.
"""
self._metadata[key] = value
@dataclass(frozen=True)
class MiddlewareResponse:
"""Immutable response object returned by middleware or terminal handler."""
status_code: int = 200
headers: dict[str, str] = field(default_factory=dict)
body: bytes | None = None
def with_status(self, code: int) -> "MiddlewareResponse":
"""Return a new response with the given status code."""
return MiddlewareResponse(status_code=code, headers=self.headers, body=self.body)
def with_body(self, body: bytes) -> "MiddlewareResponse":
"""Return a new response with the given body."""
return MiddlewareResponse(
status_code=self.status_code,
headers=self.headers,
body=body,
)
# Public interface that middleware authors implement
class MiddlewareHandler(Protocol):
"""Contract for all middleware handlers in the pipeline.
Signature: (context, next) -> response
- context: The request context with attached metadata
- next: A callable that invokes the next middleware or terminal handler
- returns: A response object (or None to continue without producing a response)
The pipeline executes handlers in priority order (lowest number first).
Each handler can short-circuit by returning a response before calling next().
"""
def __call__(self, context: MiddlewareContext, next_fn: Callable[[MiddlewareContext], MiddlewareResponse]) -> MiddlewareResponse: ...
class PriorityOrderedMiddleware(ABC):
"""Abstract base that middleware authors SHOULD subclass for priority ordering.
Override the 'priority' property to control execution order within your
middleware category. Lower numbers execute first. Default priority is 100.
"""
@property
def priority(self) -> int:
return 100
@abstractmethod
async def __call__(self, context: MiddlewareContext, next_fn: Callable[[MiddlewareContext], MiddlewareResponse]) -> MiddlewareResponse:
...
# Pipeline engine (internal framework component)
class MiddlewarePipeline:
"""Composable middleware pipeline with priority ordering and short-circuit support.
Usage by framework bootstrap code:
pipeline = MiddlewarePipeline()
pipeline.add(CacheMiddleware()) # priority=10
pipeline.add(AuthMiddleware()) # priority=20
pipeline.add(RateLimitMiddleware()) # priority=50
pipeline.add(MyPluginMiddleware()) # priority=100 (default)
response = await pipeline.handle(context) # Executes chain
"""
def __init__(self) -> None:
self._handlers: list[tuple[int, MiddlewareHandler]] = []
def add(self, handler: MiddlewareHandler | PriorityOrderedMiddleware, priority: int = 100) -> None:
"""Add a middleware handler to the pipeline.
Args:
handler: The middleware instance. If it has a 'priority' attribute,
that value is used; otherwise the explicit priority arg is used.
priority: Execution order (lower = first). Defaults to 100.
"""
if hasattr(handler, "priority"):
actual_priority = handler.priority
else:
actual_priority = priority
self._handlers.append((actual_priority, handler))
# Re-sort after insertion (small list, O(n log n) is fine)
self._handlers.sort(key=lambda x: x[0])
async def handle(self, context: MiddlewareContext) -> MiddlewareResponse:
"""Execute the full middleware chain."""
if not self._handlers:
return MiddlewareResponse(status_code=503, body=b"Service unavailable")
# Build recursive chain — each handler receives a 'next' callable
async def run(index: int) -> MiddlewareResponse:
if index >= len(self._handlers):
return MiddlewareResponse(status_code=404, body=b"Not found") # Terminal fallback
priority, handler = self._handlers[index]
try:
response = await handler(context, lambda ctx: run(index + 1))
except Exception as exc:
# Error middleware catches — logs and returns error response
return MiddlewareResponse(
status_code=500,
headers={"content-type": "application/json"},
body=f'{{"error": "{type(exc).__name__}: {exc}"}}'.encode(),
)
# Short-circuit: if handler returned a response without calling next(),
# skip remaining handlers in the chain
if response.status_code != 0 and response.body is not None:
return response
return await run(index) # Ensure we always return from terminal fallback
return await run(0)
# --- Concrete Middleware Examples ---
class AuthMiddleware(PriorityOrderedMiddleware):
"""Validates authentication tokens from request headers."""
@property
def priority(self) -> int:
return 20 # Runs after caching, before business logic
async def __call__(self, context: MiddlewareContext, next_fn: Callable[[MiddlewareContext], MiddlewareResponse]) -> MiddlewareResponse:
# PRE: Validate auth before proceeding
token = context.headers.get("authorization")
if not token:
return MiddlewareResponse(status_code=401, body=b'{"error": "unauthorized"}')
# Attach authenticated user to context metadata for downstream handlers
context.set("user_id", "user_123") # Simplified — real impl validates token
context.set("roles", ["admin", "editor"])
# Call next handler in chain
response = await next_fn(context)
# POST: Add security headers to every response
response.headers["x-authenticated"] = "true"
return response
class RateLimitMiddleware(PriorityOrderedMiddleware):
"""Throttles requests per user. Short-circuits with 429 when limit exceeded."""
@property
def priority(self) -> int:
return 50
def __init__(self, max_requests: int = 100, window_seconds: int = 60) -> None:
self._max_requests = max_requests
self._window_seconds = window_seconds
# In production: use Redis or in-memory LRU cache with TTL
self._request_counts: dict[str, list[float]] = {}
async def __call__(self, context: MiddlewareContext, next_fn: Callable[[MiddlewareContext], MiddlewareResponse]) -> MiddlewareResponse:
user_id = context.get("user_id")
if not user_id:
# Cannot rate-limit unauthenticated requests — skip for now
return await next_fn(context)
import time
now = time.time()
window_start = now - self._window_seconds
# Count requests in current window
if user_id not in self._request_counts:
self._request_counts[user_id] = []
self._request_counts[user_id] = [
t for t in self._request_counts[user_id] if t > window_start
]
count = len(self._request_counts[user_id])
if count >= self._max_requests:
return MiddlewareResponse(
status_code=429,
headers={"retry-after": str(self._window_seconds)},
body=b'{"error": "rate limit exceeded"}',
)
self._request_counts[user_id].append(now)
context.set("remaining_requests", self._max_requests - count - 1)
return await next_fn(context)
class CacheMiddleware(PriorityOrderedMiddleware):
"""Caches GET responses. Skips cache for non-GET methods."""
@property
def priority(self) -> int:
return 10 # First in chain — check cache before anything else
async def __call__(self, context: MiddlewareContext, next_fn: Callable[[MiddlewareContext], MiddlewareResponse]) -> MiddlewareResponse:
if context.method != "GET":
return await next_fn(context)
cache_key = f"cache:{context.path}"
cached = self._get_cached(cache_key) # type: ignore[return-value]
if cached is not None:
return cached.with_status(200).with_body(cached.body or b'{}')
response = await next_fn(context)
if response.status_code == 200 and response.body:
self._store_cached(cache_key, response) # type: ignore[unreachable]
return response
def _get_cached(self, key: str) -> MiddlewareResponse | None:
"""Simplified cache lookup — production would use Redis/Memcached."""
return None
def _store_cached(self, key: str, response: MiddlewareResponse) -> None:
"""Simplified cache store — production would set TTL and handle eviction."""
pass
Anti-Pattern: Leaking Internal State Through Extension Interfaces
The most common mistake in framework extension design is exposing internal implementation details through the public API. When plugin authors can read or modify internal state, any framework change becomes a breaking change for all plugins.
# ❌ BAD — Plugin interface leaks internal framework state
class LegacyPluginInterface:
"""Plugin must interact with these internal objects directly."""
def on_request(self, request_obj: Any) -> dict | None:
# Plugin receives raw internal request object with mutable state
# Framework may change the structure at any time — no versioning guarantee
request_obj._cache = {"processed_by": "my_plugin"} # Side effect on internals
if "X-API-Key" not in request_obj.headers:
return {"error": "unauthorized"} # Magic dict response, no contract
return None
def register_hook(self, framework_instance: Any) -> None:
# Plugin must hold a reference to the global framework instance
# This couples plugin code to framework internals and prevents testing
self._fw = framework_instance
self._fw._plugin_registry.append(self) # Direct mutation of internal list
# ✅ GOOD — Extension interface with explicit contracts and isolation
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class RequestContext:
"""Immutable snapshot of request data provided to plugins.
Plugins may read any field but MUST NOT modify it — the framework
creates a fresh instance per request. This ensures plugins cannot
corrupt state or create cross-request interference.
"""
method: str
path: str
headers: frozenset[tuple[str, str]] # Immutable, hashable for caching
def get_header(self, name: str) -> str | None:
"""Case-insensitive header lookup."""
lower = name.lower()
return next((v for k, v in self.headers if k.lower() == lower), None)
@dataclass(frozen=True)
class PluginResult:
"""Return value contract — plugins MUST return this type.
The framework recognizes only these three outcomes:
- proceed(): continue to the next handler or the default behavior
- halt(reason): short-circuit with a reason string (logged at WARN level)
- rewrite(context): replace the request context (rare, requires explicit enable)
"""
decision: str = "proceed" # "proceed", "halt", "rewrite"
reason: str | None = None
new_context: RequestContext | None = None
class PluginInterface(ABC):
"""Stable extension contract — plugin authors implement this interface.
Once published in a minor version of the framework, methods, parameters,
and return types defined here CANNOT change without a major version bump.
Each method receives only the data explicitly provided by the framework.
No internal objects, no global state references, no mutable shared data.
"""
@abstractmethod
def on_request_start(self, context: RequestContext) -> PluginResult:
"""Hook fired immediately after request parsing, before routing.
Args:
context: Immutable snapshot of the incoming request. Plugins
may read headers, method, and path but must not modify.
Returns:
PluginResult indicating whether to proceed, halt, or rewrite
the request context. A return value of None is treated as
'proceed' for backward compatibility with older plugin versions.
"""
...
@abstractmethod
def on_request_end(self, context: RequestContext, status_code: int) -> PluginResult:
"""Hook fired after response generation, before sending to client.
Use this to attach headers (e.g., X-Plugin-Processed), log metrics,
or short-circuit the response in error conditions.
Args:
context: The request that was processed.
status_code: The HTTP status code determined by the handler.
Returns:
PluginResult — typically 'proceed' (default). Use 'rewrite'
only if you need to replace the response entirely.
"""
...
Key differences:
- ❌ BAD: Receives raw
Anytypes — compiler can't catch mistakes, framework changes silently break plugins - ✅ GOOD: Explicit frozen dataclasses with typed signatures — IDE autocomplete, type-checking, immutable by design
- ❌ BAD: Direct mutation of internal lists (
_plugin_registry.append) — race conditions, no encapsulation - ✅ GOOD: Framework controls registration; plugin declares intent through method signatures only
Pattern 3: Versioned Extension API with Deprecation Warnings
Extension APIs evolve. This pattern shows how to version interfaces gracefully using deprecation warnings, backward-compatible additions, and clear migration paths. The pattern uses Python but the same principles apply to every language's type system.
# framework/extensions/v1/contracts.py — V1 public API (deprecated in v2)
from __future__ import annotations
import warnings
from abc import ABC, abstractmethod
from typing import Any
class PluginV1(ABC):
"""V1 plugin interface — DEPRECATED. Use PluginV2 instead.
Deprecation Notice: This interface will be removed in framework version 4.0.
Migration Guide: https://docs.example.com/migrations/plugin-v1-to-v2
"""
@property
def name(self) -> str:
return "unnamed"
@property
def version(self) -> str:
return "0.0.0"
def handle_event(self, event_name: str, payload: dict[str, Any]) -> Any:
"""Handle a framework event.
WARNING: This method receives raw dicts instead of typed events.
V2 uses strongly-typed Event objects for better type safety.
"""
warnings.warn(
"PluginV1.handle_event is deprecated since framework 3.0. "
"Use PluginV2.handle_event with typed Event parameter.",
DeprecationWarning,
stacklevel=2,
)
return payload
# framework/extensions/v2/contracts.py — V2 public API (current stable)
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Any, Protocol
class EventType(Enum):
"""Typed event types that the framework emits to plugins."""
REQUEST_STARTED = "request.started"
REQUEST_COMPLETED = "request.completed"
REQUEST_ERROR = "request.error"
SHUTDOWN_INITIATED = "system.shutdown"
@dataclass(frozen=True)
class Event:
"""Immutable event object passed to plugin handlers.
Unlike V1's raw dicts, this provides type-safe access to event data
and prevents plugins from mutating framework state accidentally.
"""
type: EventType
timestamp: datetime = None # type: ignore[assignment]
payload: dict[str, Any] = None # type: ignore[assignment]
def __post_init__(self) -> None:
if self.timestamp is None:
object.__setattr__(self, "timestamp", datetime.utcnow())
if self.payload is None:
object.__setattr__(self, "payload", {})
class LifecycleManagerV2(Protocol):
"""V2 lifecycle manager — provides typed access to framework capabilities."""
def register_shutdown_hook(self, callback: Any) -> None: ...
def get_config(self, key: str) -> Any: ...
def emit_event(self, event_name: str | EventType, payload: dict[str, Any] = None) -> None: ...
class PluginV2(ABC):
"""V2 plugin interface — the current stable API.
Changes from V1:
- handle_event receives typed Event objects instead of dicts
- LifecycleManager is injected via constructor parameter
- New on_configure() hook for reading plugin-specific config at startup
"""
@property
@abstractmethod
def name(self) -> str: ...
@property
@abstractmethod
def version(self) -> str: ...
@property
def supported_event_types(self) -> list[EventType]:
"""Which event types this plugin handles. Override to restrict processing."""
return list(EventType) # Default: handle all events
@abstractmethod
def __init__(self, manager: LifecycleManagerV2) -> None: ...
def on_configure(self, manager: LifecycleManagerV2) -> None:
"""Called during plugin activation, before event handling begins.
Use this to read plugin-specific configuration and perform one-time setup.
V1 required doing this in handle_event, which was error-prone.
"""
pass
def handle_event(self, manager: LifecycleManagerV2, event: Event) -> Any:
"""Handle a typed framework event.
Args:
manager: Lifecycle manager for framework interaction
event: Typed event object with known structure based on EventType
Returns:
Optional modification to the event payload (None = no change)
"""
if event.type not in self.supported_event_types:
return None
return None
# --- Backward Compatibility Adapter (internal, NOT public API) ---
class V1ToV2Adapter(PluginV2):
"""Internal adapter that wraps V1 plugins for the V2 pipeline.
This allows existing V1 plugins to work with framework v3.x without
requiring immediate migration. New plugin development MUST use V2 directly.
"""
def __init__(self, legacy_plugin: PluginV1) -> None:
self._legacy = legacy_plugin
# Auto-generate metadata from V1 plugin
self.name = legacy_plugin.name # type: ignore[attr-defined]
self.version = legacy_plugin.version # type: ignore[attr-defined]
@property
def supported_event_types(self) -> list[EventType]:
return list(EventType)
def handle_event(self, manager: LifecycleManagerV2, event: Event) -> Any:
# Convert V2 Event → V1 dict format for legacy plugin
return self._legacy.handle_event(event.type.value, event.payload) # type: ignore[attr-defined]
# --- Migration guide code snippet (for documentation) ---
"""
MIGRATION GUIDE: PluginV1 → PluginV2
Step 1: Change base class
BEFORE: class MyPlugin(PluginV1):
AFTER: class MyPlugin(PluginV2):
Step 2: Accept LifecycleManager in constructor
BEFORE: def __init__(self) -> None: ...
AFTER: def __init__(self, manager: LifecycleManagerV2) -> None:
self._manager = manager
Step 3: Migrate handle_event signature
BEFORE: def handle_event(self, event_name: str, payload: dict[str, Any]) -> Any:
if event_name == "request.started": ...
AFTER: def handle_event(self, manager: LifecycleManagerV2, event: Event) -> Any:
if event.type == EventType.REQUEST_STARTED: ...
Step 4: Move setup from handle_event to on_configure
BEFORE (inside handle_event): self._cache = RedisClient(...)
AFTER (new method):
def on_configure(self, manager: LifecycleManagerV2) -> None:
self._cache = RedisClient()
"""
# --- Example V2 Plugin ---
class MetricsPlugin(PluginV2):
"""Collects request metrics using the V2 API."""
@property
def name(self) -> str:
return "metrics-collector"
@property
def version(self) -> str:
return "2.1.0"
@property
def supported_event_types(self) -> list[EventType]:
return [EventType.REQUEST_STARTED, EventType.REQUEST_COMPLETED]
def __init__(self, manager: LifecycleManagerV2) -> None:
self._manager = manager
self._request_counts: dict[str, int] = {"started": 0, "completed": 0}
def handle_event(self, manager: LifecycleManagerV2, event: Event) -> Any:
if event.type == EventType.REQUEST_STARTED:
self._request_counts["started"] += 1
elif event.type == EventType.REQUEST_COMPLETED:
self._request_counts["completed"] += 1
# Emit custom metric via framework telemetry system
manager.emit_event("custom.metric", {
"name": "framework.requests_total",
"value": self._request_counts["completed"],
})
return None
# --- Version compatibility checker (internal) ---
def validate_plugin_compatibility(
plugin: PluginV2,
framework_version: str,
) -> list[str]:
"""Check if a plugin is compatible with the current framework version.
Returns a list of issues (empty = fully compatible).
Each issue describes what needs to change.
"""
issues: list[str] = []
# Check major version alignment
plugin_major = int(plugin.version.split(".")[0]) if "." in plugin.version else 0
framework_major = int(framework_version.split(".")[0]) if "." in framework_version else 0
if plugin_major < framework_major - 1:
issues.append(
f"Plugin {plugin.name} v{plugin.version} requires migration "
f"from framework v{plugin_major}.x to v{framework_major}.x. "
f"See migration guide at https://docs.example.com/migrations/"
)
return issues
Pattern 4: Extension Point Discovery and Registration System
Plugins need a reliable way to be discovered, validated, and registered. This pattern shows a plugin registry that supports file-system discovery, metadata validation, dependency resolution, and hot-reload for development mode.
# framework/extensions/discovery.py — Plugin discovery and registration engine
from __future__ import annotations
import importlib
import json
import pkgutil
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
@dataclass
class PluginSpec:
"""Metadata about a discovered plugin, extracted from its entry point file."""
name: str
version: str
description: str = ""
requires: list[str] = field(default_factory=list) # Dependency plugin names
priority: int = 100
module_path: str = "" # e.g., "my_plugin.core.MyPlugin"
file_path: Path = field(default_factory=Path)
@classmethod
def from_metadata_file(cls, metadata_path: Path) -> PluginSpec:
"""Load plugin spec from a standard metadata.json in the plugin directory."""
if not metadata_path.exists():
raise ValueError(f"Missing required metadata file: {metadata_path}")
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
return cls(
name=metadata["name"],
version=metadata["version"],
description=metadata.get("description", ""),
requires=metadata.get("require
…(truncated)