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.

Medium Python 3.9+ Aug 9, 2026 System design patterns 13 views 0 copies

Python code

62 lines
Python 3.9+
import 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

stdout
{
  "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

  1. Use `asyncio.run()` instead of manually creating and closing an event loop.
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from System design patterns

Related tutorials and quizzes for this topic.