How to Build a Scalable Plugin Architecture in Python with Real Todo Data

Q: Can I load plugins from a directory that users add at runtime?

Yes. Walk the directory with Path.iterdir() , import each .py file via importlib , and the decorator will register the plugin automatically. Remember to catch ImportError and log a clear message.

Q: Do I need to type‑check the JSON shape?

At a minimum, verify that each item is a dict containing the expected keys. The script includes a simple schema check; for stricter contracts you could use pydantic or TypedDict .

You've built a simple Todo API, but every new requirement forces you to edit the main loop—adding filters, enrichments, or exporters one by one. The answer is to move those steps into independent plugins that register themselves automatically. With a self‑registering registry the core stays untouched, test suites stay focused, and production pipelines stay maintainable. Below I show the pattern on the public JSONPlaceholder Todo feed, demonstrate two real plugins, and expose the edge cases you must guard against in production.

Key Takeaways

Self‑registration via a decorator eliminates the need to modify core code when adding new processing steps.
Validating the HTTP response (status, timeout, schema) prevents silent pipeline failures.
The registry pattern scales: you can drop dozens of plugins into a plugins/ package and the pipeline will discover them automatically.

Data and Sources

The example uses the public JSONPlaceholder Todos endpoint, which returns a stable array of 200 objects. Each object contains userId, id, title, and completed. The data was accessed on 2026-11-14.

Source: https://jsonplaceholder.typicode.com/todos

Designing a Plugin Architecture

Core concepts

The pipeline consists of three parts:

Fetcher – a thin wrapper around requests.get that enforces a timeout and checks response.status_code == 200.
Plugin registry – a global dictionary populated by a @register_plugin decorator. Each plugin implements a process(self, todos: list[dict]) -> list[dict] method.
Orchestrator – iterates over the registered plugins in the order they were defined, prints a concise status line, and finally writes the result to processed_todos.json.

Why a decorator?

In our recent post on metaprogramming decorators we saw how decorators can inject behaviour at import time. The same mechanism lets a plugin announce itself the moment its module is imported, meaning the core never needs to know the plugin's name.

Example plugins

Two concrete plugins illustrate typical Todo processing steps.

FilterIncomplete – keeps only items where completed == false. With the placeholder data this reduces the count from 200 to roughly 100.
UppercaseTitle – transforms every title to upper‑case, useful when downstream systems expect a normalized case.

Putting it together

The orchestrator loads plugins via importlib.import_module. Because each plugin module runs its decorator at import time, the registry is fully populated before the first item is processed.

Limitations and Tradeoffs

The pattern shines in medium‑scale pipelines (dozens of plugins) but has clear boundaries:

Import‑time side effects – If a plugin performs heavy I/O at import, start‑up latency grows. Keep registration lightweight.
Order sensitivity – Plugins run in registration order. When order matters, either encode a priority attribute or document the required sequence.
Runtime introspection – Adding plugins after the orchestrator has started requires a reload mechanism, which adds complexity and is outside the scope of this post.
Testing – Because plugins register globally, tests that import different plugin sets can interfere. Isolate tests by clearing the registry between runs.

Frequently Asked Questions

Can I load plugins from a directory that users add at runtime?

Yes. Walk the directory with Path.iterdir(), import each .py file via importlib, and the decorator will register the plugin automatically. Remember to catch ImportError and log a clear message.

What happens if a plugin raises an exception?

The orchestrator catches any exception, logs the plugin name and the error, and aborts the pipeline. This fail‑fast behavior prevents downstream corruption and makes the failure obvious in logs.

Do I need to type‑check the JSON shape?

At a minimum, verify that each item is a dict containing the expected keys. The script includes a simple schema check; for stricter contracts you could use pydantic or TypedDict.

Is the registry thread‑safe?

In the single‑process, synchronous pipeline shown here, thread safety is not required. If you move to a multi‑threaded executor, protect the registry with a threading.Lock or switch to an immutable registration phase.

How does this differ from a simple list of callables?

A list works for a handful of steps, but a registry gives you discoverability (e.g., registry.keys()), self‑documentation, and the ability to load plugins lazily from external packages.

Complete Script

import json
import logging
import sys
from pathlib import Path
from typing import Callable, Dict, List, Protocol

import requests

# ----------------------------------------------------------------------
# Logging configuration
# ----------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    stream=sys.stdout,
)

logger = logging.getLogger(__name__)

# ----------------------------------------------------------------------
# Plugin protocol and registry
# ----------------------------------------------------------------------
class TodoProcessor(Protocol):
    """A callable that receives a list of todo dicts and returns a new list."""
    def process(self, todos: List[Dict]) -> List[Dict]:
        ...

_plugin_registry: Dict[str, TodoProcessor] = {}

def register_plugin(name: str) -> Callable[[TodoProcessor], TodoProcessor]:
    """Decorator that registers a plugin class under *name*."""
    def decorator(cls: TodoProcessor) -> TodoProcessor:
        if name in _plugin_registry:
            raise ValueError(f"Plugin '{name}' already registered")
        _plugin_registry[name] = cls()
        logger.debug("Registered plugin %s", name)
        return cls
    return decorator

# ----------------------------------------------------------------------
# Example plugins (would normally live in separate modules)
# ----------------------------------------------------------------------
@register_plugin("FilterIncomplete")
class FilterIncomplete:
    def process(self, todos: List[Dict]) -> List[Dict]:
        filtered = [t for t in todos if not t.get("completed", True)]
        return filtered

@register_plugin("UppercaseTitle")
class UppercaseTitle:
    def process(self, todos: List[Dict]) -> List[Dict]:
        for t in todos:
            if "title" in t:
                t["title"] = t["title"].upper()
        return todos

# ----------------------------------------------------------------------
# Core fetcher with error handling
# ----------------------------------------------------------------------
def fetch_todos(url: str, timeout: int = 5) -> List[Dict]:
    try:
        response = requests.get(url, timeout=timeout)
    except requests.RequestException as exc:
        logger.error("Network error while fetching todos: %s", exc)
        raise SystemExit(1) from exc

    if response.status_code != 200:
        logger.error("Unexpected status %s from %s", response.status_code, url)
        raise SystemExit(1)

    try:
        data = response.json()
    except json.JSONDecodeError as exc:
        logger.error("Failed to decode JSON: %s", exc)
        raise SystemExit(1) from exc

    if not isinstance(data, list):
        logger.error("Expected a list of todos, got %s", type(data).__name__)
        raise SystemExit(1)

    # Minimal schema validation
    required_keys = {"userId", "id", "title", "completed"}
    for i, item in enumerate(data):
        if not isinstance(item, dict) or not required_keys.issubset(item):
            logger.error("Todo at index %d is malformed: %s", i, item)
            raise SystemExit(1)

    logger.info("Fetched %d todos", len(data))
    return data

# ----------------------------------------------------------------------
# Orchestrator that runs the registered plugins
# ----------------------------------------------------------------------
def run_pipeline(todos: List[Dict]) -> List[Dict]:
    current = todos
    for name, plugin in _plugin_registry.items():
        try:
            current = plugin.process(current)
            logger.info("After %s: %d items", name, len(current))
        except Exception as exc:
            logger.exception("Plugin %s failed: %s", name, exc)
            raise SystemExit(1) from exc
    return current

# ----------------------------------------------------------------------
# Main entry point
# ----------------------------------------------------------------------
def main() -> None:
    url = "https://jsonplaceholder.typicode.com/todos"
    raw_todos = fetch_todos(url)

    processed = run_pipeline(raw_todos)

    output_path = Path("processed_todos.json")
    try:
        output_path.write_text(json.dumps(processed, indent=2))
        logger.info("Wrote %d processed todos to %s", len(processed), output_path)
    except OSError as exc:
        logger.error("Failed to write output file: %s", exc)
        raise SystemExit(1) from exc

if __name__ == "__main__":
    main()

Recommendation and Next Step

If you already have a Todo service, replace the monolithic loop with the script above, add your domain‑specific plugins to a plugins/ package, and run the pipeline as part of your nightly ETL or CI validation. The next step is to containerise the script, add health‑check endpoints, and configure a retry policy in your orchestrator (Airflow, Prefect, etc.) so that transient network failures are handled automatically.

Py Data | Python, Data Engineering & AI Tutorials