Microkernel Plug-in Core Mock in Python
Implements a minimal microkernel plug-in core that registers, unregisters, and executes synchronous or asynchronous plugins via a pluggable manager class.
Python code
62 linesimport json
import abc
import inspect
class MicrokernelCore(abc.ABC):
def __init__(self):
self._plugins = {}
def register(self, name, plugin):
self._plugins[name] = plugin
def unregister(self, name):
return self._plugins.pop(name, None)
def execute(self, name, *args, **kwargs):
if name not in self._plugins:
raise KeyError(f"Plugin '{name}' is not registered")
plugin = self._plugins[name]
return self._execute(plugin, *args, **kwargs)
@abc.abstractmethod
def _execute(self, plugin, *args, **kwargs):
pass
class PluginManager(MicrokernelCore):
def _execute(self, plugin, *args, **kwargs):
if inspect.iscoroutinefunction(plugin):
import asyncio
loop = asyncio.new_event_loop()
try:
return loop.run_until_complete(plugin(*args, **kwargs))
finally:
loop.close()
return plugin(*args, **kwargs)
def list_plugins(self):
return list(self._plugins.keys())
if __name__ == "__main__":
def add(a, b):
return a + b
async def multiply(a, b):
return a * b
core = PluginManager()
core.register("add", add)
core.register("multiply_sync", lambda a, b: a * b)
core.register("multiply_async", multiply)
print(json.dumps({
"plugins": core.list_plugins(),
"add_result": core.execute("add", 2, 3),
"multiply_sync_result": core.execute("multiply_sync", 2, 3),
"multiply_async_result": core.execute("multiply_async", 2, 3),
}, indent=2))
Output
{
"plugins": [
"add",
"multiply_sync",
"multiply_async"
],
"add_result": 5,
"multiply_sync_result": 6,
"multiply_async_result": 6
}
How it works
The MicrokernelCore abstract base class defines the contract for plugin registration, unregistration, and execution, forcing subclasses to implement the _execute method. The PluginManager fills in the execution logic, using inspect.iscoroutinefunction to detect async plugins and running them on a fresh event loop. This decouples core kernel logic from plugin specifics, allowing plugins to be added or removed at runtime without modifying the core. The list_plugins method provides visibility into the registered plugins, and the __main__ block demonstrates both synchronous and asynchronous plugins in a single output.
Common mistakes
- Forgetting to implement `_execute` in subclasses, causing `TypeError` at runtime.
- Not closing the event loop after running an async plugin, leaking resources.
- Assuming plugin names are case-sensitive when they need to be handled consistently.
- Calling `execute` with a plugin name that doesn't exist without checking `list_plugins` first.
Variations
- Use `asyncio.run()` instead of manually creating and closing an event loop.
- Add plugin metadata (e.g., version, description) to the registry by storing dicts instead of plain functions.
Real-world use cases
- Building an extensible plugin system for a data processing pipeline where each plugin can be sync or async.
- Creating a microservice core that loads feature modules at runtime without redeploying the main application.
- Implementing a task executor where plugins represent different strategies for handling user requests in a web framework.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.