Home
architecture
python

How to Build a Robust Plugin Architecture in Python Using Real RSS Data

Mausam

You're running an RSS ingestion pipeline that hard‑codes every transformation, so every new business need forces you to edit core code, risk regressions, and redeploy the whole service. The solution is to refactor the pipeline into a lightweight plugin system: each processing step lives in its own plug‑in, the core loader discovers and runs them automatically, and you can enable, replace, or turn off steps without touching the main script. Below I demonstrate this pattern with the public Discord Engineering RSS feed and a runnable script that proves the concept.

Key Takeaways

  • Decoupling transformations into plug‑ins isolates change and eliminates cascade failures when requirements evolve.
  • Dynamic discovery (via a registry or import‑based scanning) lets you add new logic without restarting or rebuilding the core service.
  • Explicit error handling around network fetches and plug‑in contracts keeps the pipeline robust in production.

Data and Sources

The demonstration uses the Discord Engineering RSS feed, a stable public XML source that updates daily.

Source reviewed on 2026-11-07.

Designing a Plugin Architecture in Python

At its core, a plug‑in system needs three pieces:

  1. Contract. A base class or abstract protocol that defines the process(entry) method each plug‑in must implement.
  2. Discovery. A mechanism that scans a known package (or a registry) and loads every concrete plug‑in class.
  3. Execution engine. The core loop that fetches the RSS feed, validates it, and hands each entry to every enabled plug‑in.

Because the pipeline already exists from the “Autonomous Data Pipelines” post, we keep the same main() entry point and simply inject the plug‑in step list.

Implementing Plugin Discovery and Execution

The script below follows a pragmatic approach that works in any standard Python environment:

  • Plug‑ins register themselves via a @register_plugin decorator, populating a global PLUGIN_REGISTRY.
  • The core loader imports the plugins module, which triggers registration of all plug‑in classes defined there.
  • A simple config.yaml (embedded as a dict for brevity) lists enabled plug‑ins, allowing you to toggle them without code changes.
  • Network fetch uses requests with a timeout and validates HTTP status and entry shape before handing data to plug‑ins.

Two example plug‑ins illustrate common tasks:

  • TitlePrinter – prints the title and link of each entry (the baseline behavior).
  • AuthorExtractor – extracts and logs the author name when present, demonstrating how to add richer metadata handling.

Limitations and Tradeoffs

While plug‑in architectures increase flexibility, they introduce new considerations:

  • Discovery overhead. Importing many plug‑in modules adds start‑up latency; in a high‑throughput service you may want to cache the plug‑in list.
  • Version compatibility. Independent plug‑in packages must agree on the contract; breaking changes require coordinated releases or a versioned interface.
  • Performance impact. Each plug‑in runs sequentially per entry. If you have dozens of heavy plug‑ins, consider parallel execution or batch processing.
  • State sharing. The simple contract is stateless; stateful transformations (e.g., aggregations) need a shared context object passed to process, which adds complexity.

Frequently Asked Questions

Can I load plug‑ins from external wheels?

Yes. Replace the registry decorator with importlib.metadata.entry_points() and declare an entry point group (e.g., rss_plugins) in each wheel’s setup.cfg. The core loader then iterates over those entry points.

How do I test a single plug‑in in isolation?

Instantiate the plug‑in class directly and feed it a mock feedparser.FeedParserDict entry. Because the contract is a single method, unit tests are straightforward and do not require the full RSS fetch.

What happens if a plug‑in raises an exception?

The execution engine catches any exception, logs the plug‑in name and traceback, and continues with the next plug‑in. This “fail‑fast‑but‑continue” pattern prevents one buggy plug‑in from halting the entire pipeline.

Is the architecture suitable for streaming large feeds?

For very large feeds you might want to process entries lazily (e.g., using a generator) and optionally run plug‑ins in separate worker threads or processes. The current script processes the whole list in memory for simplicity.

Do I need a database to persist plug‑in results?

Persistence is orthogonal to the plug‑in system. Each plug‑in can decide its own storage strategy—write to a file, push to a database, or emit to a message queue—without the core pipeline needing to know.

Complete Script

#!/usr/bin/env python3
"""
Robust RSS plug‑in pipeline using the Discord Engineering feed.

Key ideas demonstrated:
* Contract‑based plug‑in base class.
* Registry‑decorator discovery.
* Config‑driven enable/disable.
* Safe network fetch with timeout and shape validation.
* Two example plug‑ins (TitlePrinter, AuthorExtractor).
"""

import importlib
import logging
import sys
from pathlib import Path
from typing import List, Protocol, runtime_checkable

import requests
import feedparser

# ----------------------------------------------------------------------
# Configuration (could be external YAML/JSON; kept inline for brevity)
# ----------------------------------------------------------------------
CONFIG = {
    "feed_url": "https://discord.com/blog/rss.xml",
    "request_timeout": 5,               # seconds
    "enabled_plugins": [
        "TitlePrinter",
        "AuthorExtractor",
    ],
}

# ----------------------------------------------------------------------
# Logging setup
# ----------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s - %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger("rss_plugin_pipeline")

# ----------------------------------------------------------------------
# Plug‑in contract
# ----------------------------------------------------------------------
@runtime_checkable
class RSSPlugin(Protocol):
    """A plug‑in must implement a single `process(entry)` method."""
    def process(self, entry: feedparser.FeedParserDict) -> None:
        ...

# ----------------------------------------------------------------------
# Registry for plug‑ins
# ----------------------------------------------------------------------
PLUGIN_REGISTRY = {}

def register_plugin(cls):
    """Class decorator that registers a concrete plug‑in."""
    if not issubclass(cls, RSSPlugin):
        raise TypeError(f"{cls.__name__} does not implement RSSPlugin protocol")
    PLUGIN_REGISTRY[cls.__name__] = cls
    return cls

# ----------------------------------------------------------------------
# Example plug‑ins (normally live in a separate `plugins` package)
# ----------------------------------------------------------------------
@register_plugin
class TitlePrinter:
    """Prints title and link of each entry."""
    def process(self, entry):
        title = entry.get("title", "No Title")
        link = entry.get("link", "No Link")
        log.info("TitlePrinter → %s (%s)", title, link)

@register_plugin
class AuthorExtractor:
    """Extracts author information if present."""
    def process(self, entry):
        author = entry.get("author")
        if author:
            log.info("AuthorExtractor → author: %s", author)
        else:
            log.debug("AuthorExtractor → no author field for entry %s", entry.get("id", "unknown"))

# ----------------------------------------------------------------------
# Core functions
# ----------------------------------------------------------------------
def fetch_feed(url: str, timeout: int) -> feedparser.FeedParserDict:
    """Fetch RSS XML with a timeout and return a parsed feed."""
    try:
        response = requests.get(url, timeout=timeout, headers={"User-Agent": "rss-plugin-demo/1.0"})
        response.raise_for_status()
    except requests.RequestException as exc:
        log.error("Failed to fetch RSS feed: %s", exc)
        raise RuntimeError("Network error while retrieving RSS feed") from exc

    feed = feedparser.parse(response.content)
    if feed.bozo:
        log.error("Malformed RSS feed: %s", feed.bozo_exception)
        raise ValueError("Feed parsing failed")
    if not getattr(feed, "entries", None):
        raise ValueError("Feed contains no entries")
    return feed

def load_plugins(enabled: List[str]) -> List[RSSPlugin]:
    """Instantiate enabled plug‑ins from the registry."""
    plugins = []
    for name in enabled:
        cls = PLUGIN_REGISTRY.get(name)
        if not cls:
            log.warning("Requested plug‑in %s not found in registry; skipping", name)
            continue
        plugins.append(cls())
        log.debug("Loaded plug‑in %s", name)
    return plugins

def run_pipeline():
    """Main orchestration: fetch feed, load plug‑ins, process entries."""
    feed = fetch_feed(CONFIG["feed_url"], CONFIG["request_timeout"])
    plugins = load_plugins(CONFIG["enabled_plugins"])

    if not plugins:
        log.warning("No plug‑ins loaded; exiting pipeline")
        return

    entry_count = 0
    for entry in feed.entries:
        entry_count += 1
        for plugin in plugins:
            try:
                plugin.process(entry)
            except Exception as exc:
                log.exception("Plug‑in %s raised an exception on entry %s: %s",
                              plugin.__class__.__name__, entry.get("id", "unknown"), exc)
                # Continue with next plug‑in
    log.info("Processed %d entries with %d plug‑ins", entry_count, len(plugins))

def main():
    """Entry point."""
    try:
        run_pipeline()
    except Exception as exc:
        log.critical("Pipeline terminated unexpectedly: %s", exc)
        sys.exit(1)

if __name__ == "__main__":
    main()

Recommendation and Next Step

Adopt the plug‑in skeleton above for any existing RSS or API ingestion pipeline: start by extracting one transformation (e.g., title printing) into its own plug‑in, verify that the core still runs, then iteratively migrate remaining steps. Once the architecture proves stable, publish each plug‑in as an independent wheel so teams can version, test, and deploy them separately without touching the core loader.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.