Ever found yourself scrambling to add a new data‑source to a script that’s already a tangled mess of if‑else blocks? I hit that wall last month while extending the multi‑agent LLM trading bot from my earlier post – every new market feed meant editing the same 300‑line file, breaking tests, and triggering a cascade of merge conflicts. What if you could drop a new “processor” into a folder and have the system magically pick it up? In this tutorial I’ll show you exactly how I turned a monolithic Todo‑processor into a plug‑and‑play platform using Python’s standard library, and why that pattern is now the backbone of every new feature in my production pipelines.
Key Takeaways
- A well‑defined abstract base class (ABC) creates an immutable contract that every plugin must obey.
- Dynamic discovery with
importliblets you load plugins from a directory at runtime, keeping the core code untouched. - Isolating plugin failures with try/except and optional process isolation prevents a single bad plugin from crashing the whole service.
- Bootstrapping a minimal plugin folder on first run makes the system self‑documenting and ready for contributors.
- For production, pair the plugin host with logging, health checks, and versioned config to avoid silent regressions.
The Problem
When I first built the Todo‑pipeline (see Async/await in Production), all transformations lived in process_todos(). Adding a new filter meant opening that file, inserting code, and re‑running the entire test suite. The pain grew as the team added LLM agents, statistical enrichers, and external risk checks. The monolith became a bottleneck for iteration and a source of fragile bugs.
Data and Sources
We’ll work with the public JSONPlaceholder Todos endpoint: https://jsonplaceholder.typicode.com/todos. The dataset consists of 200 todo items, each with userId, id, title, and completed. Data accessed on 2026-11-02.
Step 1 — The Monolithic Trap: When Hardcoding Becomes a Bottleneck
In the original script I wrote a single function that filtered completed items, counted per user, and wrote CSVs. Every new requirement forced me to edit that function, increasing the risk of regression. The code looked like this:
def process_todos(data):
# filter completed
pending = [t for t in data if not t["completed"]]
# count per user
counts = {}
for t in pending:
counts[t["userId"]] = counts.get(t["userId"], 0) + 1
# write CSV
with open("out.csv", "w") as f:
f.write("userId,count\n")
for uid, cnt in counts.items():
f.write(f"{uid},{cnt}\\n")
Adding a new step—say, flagging overdue items—meant touching this block again. The solution is to extract each step into an independent plugin.
Step 2 — Defining the Contract: Crafting the Plugin Interface with ABCs
The contract guarantees that the host can call any plugin in the same way. I used abc.ABC to enforce the process method signature.
from abc import ABC, abstractmethod
from typing import List, Dict, Any
class PluginBase(ABC):
"""All plugins must inherit from this class."""
@abstractmethod
def process(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Transform the incoming list of todo dicts and return a new list."""
pass
Because the method returns a list of dictionaries, the host can chain plugins without caring about their internal logic.
Step 3 — Dynamic Discovery and Loading: importlib in Action
The host walks a configurable directory (plugins/) and imports every .py file that defines a subclass of PluginBase. Errors while importing a plugin should never crash the whole app, so each import is wrapped in a try/except.
import importlib.util
import pathlib
import logging
logger = logging.getLogger(__name__)
def discover_plugins(plugin_dir: str) -> List[PluginBase]:
plugins = []
plugin_path = pathlib.Path(plugin_dir)
if not plugin_path.is_dir():
logger.warning("Plugin directory %s does not exist.", plugin_dir)
return plugins
for file in plugin_path.glob("*.py"):
module_name = file.stem
spec = importlib.util.spec_from_file_location(module_name, file)
try:
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore
except Exception as e:
logger.error("Failed to load plugin %s: %s", module_name, e)
continue
for attr in dir(module):
obj = getattr(module, attr)
if isinstance(obj, type) and issubclass(obj, PluginBase) and obj is not PluginBase:
try:
plugins.append(obj())
logger.info("Loaded plugin: %s", obj.__name__)
except Exception as e:
logger.error("Failed to instantiate %s: %s", obj.__name__, e)
return plugins
This function returns a list of ready‑to‑use plugin instances, ready for orchestration.
Step 4 — Building Modular Processors: Concrete Plugin Implementations
Two example plugins live in plugins/filter_completed.py and plugins/summary_stats.py. They demonstrate a pure filter and a simple aggregation.
# plugins/filter_completed.py
from plugin_base import PluginBase
class FilterCompleted(PluginBase):
def process(self, data):
return [t for t in data if not t["completed"]]
# plugins/summary_stats.py
from plugin_base import PluginBase
class SummaryStats(PluginBase):
def process(self, data):
# produce a single‑row summary per user
from collections import Counter
user_counts = Counter(t["userId"] for t in data)
return [{"userId": uid, "pending": cnt} for uid, cnt in user_counts.items()]
Both plugins inherit the contract, keep their own dependencies, and can be added or removed without touching the host.
Step 5 — The Plugin Host: Orchestrating Data Flow and Execution
The host fetches the Todo JSON, runs each plugin in sequence, and finally writes the result to output.json. If a plugin raises an exception, we log it and continue with the next plugin – a “best‑effort” strategy that works well for non‑critical enrichers.
import json
import requests
def load_data(url: str) -> List[Dict[str, Any]]:
try:
resp = requests.get(url, timeout=5)
resp.raise_for_status()
return resp.json()
except requests.exceptions.RequestException as e:
logger.error("Network error while fetching data: %s", e)
raise
def run_pipeline(data: List[Dict[str, Any]], plugins: List[PluginBase]) -> List[Dict[str, Any]]:
for plugin in plugins:
try:
data = plugin.process(data)
logger.debug("Plugin %s produced %d records", plugin.__class__.__name__, len(data))
except Exception as e:
logger.exception("Plugin %s failed: %s", plugin.__class__.__name__, e)
# continue with next plugin
return data
def save_output(data: List[Dict[str, Any]], path: str = "output.json"):
try:
with open(path, "w") as f:
json.dump(data, f, indent=2)
logger.info("Saved final output to %s", path)
except IOError as e:
logger.error("Failed to write output: %s", e)
raise
Step 6 — Robustness in Production: Isolating Plugin Failures
In a high‑throughput service you might want each plugin to run in its own process to avoid GIL contention or memory leaks. The pattern below shows a lightweight wrapper using multiprocessing that can be swapped in when needed.
import multiprocessing as mp
def run_plugin_isolated(plugin: PluginBase, data):
def _worker(inp, out_q):
try:
out_q.put(plugin.process(inp))
except Exception as exc:
out_q.put(exc)
q = mp.Queue()
p = mp.Process(target=_worker, args=(data, q))
p.start()
p.join()
result = q.get()
if isinstance(result, Exception):
raise result
return result
For the tutorial we stick to the in‑process version, but the isolated runner is ready for production scaling.
Putting It Together
Now we tie all pieces: fetch the Todo list, bootstrap a default plugins/ folder if missing, discover plugins, run the pipeline, and persist the output.
import os
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
PLUGIN_DIR = "plugins"
def bootstrap_plugins():
"""Create a minimal plugin package on first run."""
os.makedirs(PLUGIN_DIR, exist_ok=True)
# Write the base class so plugins can import it
base_path = os.path.join(PLUGIN_DIR, "plugin_base.py")
if not os.path.exists(base_path):
with open(base_path, "w") as f:
f.write("""from abc import ABC, abstractmethod
from typing import List, Dict, Any
class PluginBase(ABC):
@abstractmethod
def process(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
pass
""")
# Sample FilterCompleted plugin
fc_path = os.path.join(PLUGIN_DIR, "filter_completed.py")
if not os.path.exists(fc_path):
with open(fc_path, "w") as f:
f.write("""from plugin_base import PluginBase
class FilterCompleted(PluginBase):
def process(self, data):
return [t for t in data if not t["completed"]]
""")
# Sample SummaryStats plugin
ss_path = os.path.join(PLUGIN_DIR, "summary_stats.py")
if not os.path.exists(ss_path):
with open(ss_path, "w") as f:
f.write("""from plugin_base import PluginBase
from collections import Counter
class SummaryStats(PluginBase):
def process(self, data):
user_counts = Counter(t["userId"] for t in data)
return [{"userId": uid, "pending": cnt} for uid, cnt in user_counts.items()]
""")
def main():
bootstrap_plugins()
raw = load_data("https://jsonplaceholder.typicode.com/todos")
plugins = discover_plugins(PLUGIN_DIR)
if not plugins:
logger.warning("No plugins found – exiting.")
return
result = run_pipeline(raw, plugins)
save_output(result)
if __name__ == "__main__":
main()
Complete Script
The full runnable script combining all steps. Save it as plugin_pipeline.py and run python plugin_pipeline.py. It will create a plugins/ folder with two example plugins, fetch the Todo data, process it through the plugins, and write output.json.
#!/usr/bin/env python3
"""
Engineering Extensibility Demo:
A production‑ready plugin architecture that processes JSONPlaceholder Todos.
"""
import os
import pathlib
import logging
import json
import requests
import importlib.util
from abc import ABC, abstractmethod
from collections import Counter
from typing import List, Dict, Any
# ----------------------------------------------------------------------
# Logging configuration
# ----------------------------------------------------------------------
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
# ----------------------------------------------------------------------
# Plugin contract
# ----------------------------------------------------------------------
class PluginBase(ABC):
"""All plugins must inherit from this class."""
@abstractmethod
def process(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Transform a list of todo dicts and return a new list."""
pass
# ----------------------------------------------------------------------
# Helper: fetch data
# ----------------------------------------------------------------------
def load_data(url: str) -> List[Dict[str, Any]]:
try:
resp = requests.get(url, timeout=5)
resp.raise_for_status()
logger.info("Fetched %d records from %s", len(resp.json()), url)
return resp.json()
except requests.exceptions.RequestException as e:
logger.error("Network error while fetching data: %s", e)
raise
# ----------------------------------------------------------------------
# Plugin discovery
# ----------------------------------------------------------------------
def discover_plugins(plugin_dir: str) -> List[PluginBase]:
plugins: List[PluginBase] = []
plugin_path = pathlib.Path(plugin_dir)
if not plugin_path.is_dir():
logger.warning("Plugin directory %s does not exist.", plugin_dir)
return plugins
for file in plugin_path.glob("*.py"):
if file.name == "plugin_base.py":
continue # skip the contract itself
module_name = file.stem
spec = importlib.util.spec_from_file_location(module_name, file)
try:
module = importlib.util.module_from_spec(spec) # type: ignore
spec.loader.exec_module(module) # type: ignore
except Exception as e:
logger.error("Failed to import %s: %s", module_name, e)
continue
for attr in dir(module):
obj = getattr(module, attr)
if isinstance(obj, type) and issubclass(obj, PluginBase) and obj is not PluginBase:
try:
plugins.append(obj())
logger.info("Loaded plugin: %s", obj.__name__)
except Exception as e:
logger.error("Failed to instantiate %s: %s", obj.__name__, e)
return plugins
# ----------------------------------------------------------------------
# Pipeline runner
# ----------------------------------------------------------------------
def run_pipeline(data: List[Dict[str, Any]], plugins: List[PluginBase]) -> List[Dict[str, Any]]:
for plugin in plugins:
try:
data = plugin.process(data)
logger.debug("Plugin %s output %d rows", plugin.__class__.__name__, len(data))
except Exception as e:
logger.exception("Plugin %s crashed: %s", plugin.__class__.__name__, e)
# Continue with next plugin – best‑effort mode
return data
# ----------------------------------------------------------------------
# Persist results
# ----------------------------------------------------------------------
def save_output(data: List[Dict[str, Any]], path: str = "output.json"):
try:
with open(path, "w") as f:
json.dump(data, f, indent=2)
logger.info("Saved %d records to %s", len(data), path)
except IOError as e:
logger.error("Failed to write output file: %s", e)
raise
# ----------------------------------------------------------------------
# Bootstrap default plugins (first‑run convenience)
# ----------------------------------------------------------------------
PLUGIN_DIR = "plugins"
def bootstrap_plugins():
os.makedirs(PLUGIN_DIR, exist_ok=True)
base_path = os.path.join(PLUGIN_DIR, "plugin_base.py")
if not os.path.exists(base_path):
with open(base_path, "w") as f:
f.write(
"from abc import ABC, abstractmethod\\n"
"from typing import List, Dict, Any\\n\\n"
"class PluginBase(ABC):\\n"
" @abstractmethod\\n"
" def process(self, data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:\\n"
" pass\\n"
)
fc_path = os.path.join(PLUGIN_DIR, "filter_completed.py")
if not os.path.exists(fc_path):
with open(fc_path, "w") as f:
f.write(
"from plugin_base import PluginBase\\n\\n"
"class FilterCompleted(PluginBase):\\n"
" def process(self, data):\\n"
" return [t for t in data if not t[\\\"completed\\\"]]\\n"
)
ss_path = os.path.join(PLUGIN_DIR, "summary_stats.py")
if not os.path.exists(ss_path):
with open(ss_path, "w") as f:
f.write(
"from plugin_base import PluginBase\\n"
"from collections import Counter\\n\\n"
"class SummaryStats(PluginBase):\\n"
" def process(self, data):\\n"
" user_counts = Counter(t[\\\"userId\\\"] for t in data)\\n"
" return [{\\\"userId\\\": uid, \\\"pending\\\": cnt} for uid, cnt in user_counts.items()]\\n"
)
# ----------------------------------------------------------------------
# Entry point
# ----------------------------------------------------------------------
def main():
bootstrap_plugins()
raw = load_data("https://jsonplaceholder.typicode.com/todos")
plugins = discover_plugins(PLUGIN_DIR)
if not plugins:
logger.warning("No plugins discovered – exiting.")
return
processed = run_pipeline(raw, plugins)
save_output(processed)
if __name__ == "__main__":
main()
Expected Output
After running the script you’ll see an output.json file similar to:
[
{
"userId": 1,
"pending": 5
},
{
"userId": 2,
"pending": 7
},
...
]
The numbers correspond to the count of incomplete Todos per user, produced by the two sample plugins.
Limitations and Tradeoffs
This approach shines for loosely coupled, frequently changing processing steps. It falters when plugins need tight, low‑latency interaction (e.g., shared in‑memory models) because each plugin runs in isolation and incurs import overhead. The simple file‑system discovery also assumes trusted code – in a multi‑tenant environment you’d need signature verification, sandboxing, or containerization to avoid malicious plugins.
Frequently Asked Questions
Can I load plugins from a remote repository instead of a local folder?
Yes. Replace the filesystem walk with importlib.metadata.entry_points() or pull a zip from a private PyPI index, extract it to a temporary directory, and feed that path to discover_plugins. Just validate signatures before execution.
How do I enforce that a plugin returns a specific schema?
Wrap the return value in a Pydantic model inside the base class or add a helper method like validate_output that concrete plugins call before returning.
What if a plugin needs configuration (API keys, thresholds) that shouldn’t be hard‑coded?
Pass a configuration dictionary to each plugin’s constructor. The host can read a central config.yaml or environment variables and inject the relevant slice when instantiating the plugin.
Is there a way to run each plugin in its own process to avoid GIL contention?
Absolutely. The run_plugin_isolated helper in Step 6 demonstrates a minimal multiprocessing wrapper. For large pipelines you might use a task queue (Celery, RQ) or a lightweight process pool.
How do I test plugins without starting the whole host?
Because each plugin is a plain class with a single method, unit tests can import the module directly and feed a handcrafted list of dicts. No network or file‑system dependencies are required.
What I'd Change
If I were to ship this pattern at scale, I’d replace the ad‑hoc plugins/ folder with a versioned package registry and enforce semantic versioning on each plugin. Coupling that with a CI gate that runs contract tests against every new release would give us the confidence to let data scientists push their own processors without risking a cascade failure. In short: the core idea – a clean ABC contract + dynamic discovery – is solid, but production‑grade systems need stricter security, version control, and observability baked in from day one.