You're building a production‑grade pipeline that ingests RSS feeds, and every new content type forces you to edit a monolithic dispatch loop. The answer is to replace that hard‑coded logic with a tiny decorator that registers handler functions on import, turning your pipeline into a plug‑in‑style system that scales cleanly.
Key Takeaways
- Decorators can populate a global registry keyed by regular‑expression patterns, eliminating the need for a manual dispatcher.
- The pattern‑based registry works with any feed; the Discord Engineering RSS feed demonstrates real‑world variety (patch notes, badge releases, Nitro updates, etc.).
- Because registration happens at import time, adding a new handler is as simple as defining a function with @handler – no test‑time or runtime code changes required.
Data and Sources
The script parses the public Discord Engineering RSS feed:
- https://discord.com/blog/rss.xml (accessed via
feedparser).
Sample titles from the feed include “Discord Patch Notes: November 9, 2026”, “Unlock, Level Up, Flex: Discord Profile Badges Got an Upgrade”, and “Modernizing the Video Playback Experience at Discord”. These illustrate distinct content categories that our regex‑based handlers can target.
Source reviewed on 2026-11-11.
Decorators for Dynamic RSS Processing
The core of the pattern is a @handler decorator that receives a regular‑expression string. When the module is imported, the decorator stores the function in a global HANDLERS dict. Later, the dispatcher iterates over the feed entries, matches entry.title against the registered patterns, and calls the first matching handler.
Why this matters in production:
- Zero‑downtime updates. Deploy a new handler file, restart the worker, and the new rule is live without touching the dispatcher.
- Test isolation. Unit tests import only the handler they need, guaranteeing that registration does not leak state.
- Explicit intent. The regex lives next to the processing code, making it easy to audit.
Below is a concise example that registers three handlers for the Discord feed:
news_handler– catches titles containing “Patch Notes” or “Release”.badge_handler– catches “Badge” or “Profile”.video_handler– catches “Video Playback”.
Each handler builds a small dictionary that later gets written to categorized_entries.json. The script also prints a one‑line summary like [NEWS] – Discord Patch Notes: November 9, 2026.
Limitations and Tradeoffs
- Regex maintenance. Overly broad patterns can cause ambiguous matches; ordering of registration matters because the first match wins.
- Performance. For very large feeds (tens of thousands of entries) the linear scan of
HANDLERSfor each entry adds O(N × M) work (N = entries, M = handlers). In such cases a compiled decision tree or prefix map may be preferable. - Static analysis. Because registration happens at import time, static linters cannot warn about missing handlers for a new content type; you must rely on integration tests.
- Failure isolation. A single handler raising an exception will abort the whole dispatch unless you wrap calls in
try/except. The reference script catches generic errors and logs them, keeping the pipeline alive.
Frequently Asked Questions
Do I need to restart the worker after adding a new handler?
Yes. Registration occurs when the module is imported, so a process restart (or a dynamic reload in a long‑running service) is required to pick up new functions.
Can I use async I/O with this pattern?
Absolutely. The decorator and registry are synchronous, but the dispatcher can be wrapped in an asyncio coroutine that fetches the feed with aiohttp. The core registration logic remains unchanged.
What if an entry matches multiple patterns?
The first registered pattern wins. If you need multi‑category tagging, let handlers return a list of categories and aggregate results after dispatch.
How do I test a single handler in isolation?
Import the module, call the handler directly with a fabricated entry dict, and assert the returned payload. Because registration happens at import, you can also clear HANDLERS in test setup to avoid side effects.
Is this approach safe for multi‑process workers?
Each process maintains its own in‑memory registry, so there is no shared mutable state. The pattern scales horizontally without coordination.
Recommendation and Next Step
Adopt the decorator‑registry pattern for any plug‑in‑style ingestion layer—whether you are pulling RSS, Kafka topics, or webhook payloads. Once the registry is stable, add a thin validation layer (e.g., pydantic models) to enforce schema contracts before downstream processing. This two‑step approach gives you both extensibility and data quality guarantees.
Complete Script
import json
import re
import sys
from pathlib import Path
from typing import Callable, Dict, List
import feedparser
import requests
# Global registry: pattern (compiled regex) -> handler function
HANDLERS: Dict[re.Pattern, Callable[[dict], dict]] = {}
class MissingEntriesError(RuntimeError):
"""Raised when the fetched RSS feed contains no entries."""
def handler(pattern: str):
"""
Decorator that registers a function as a handler for titles matching *pattern*.
The pattern is compiled once and stored in the global HANDLERS dict.
"""
compiled = re.compile(pattern, re.IGNORECASE)
def wrapper(func: Callable[[dict], dict]):
if compiled in HANDLERS:
raise ValueError(f"Pattern {pattern!r} already registered")
HANDLERS[compiled] = func
return func
return wrapper
@handler(r"patch notes|release")
def news_handler(entry: dict) -> dict:
"""Process generic news or release‑note entries."""
return {"type": "NEWS", "title": entry["title"], "link": entry["link"]}
@handler(r"badge|profile")
def badge_handler(entry: dict) -> dict:
"""Process Discord badge announcements."""
return {"type": "BADGE", "title": entry["title"], "link": entry["link"]}
@handler(r"video playback")
def video_handler(entry: dict) -> dict:
"""Process posts about video playback improvements."""
return {"type": "VIDEO", "title": entry["title"], "link": entry["link"]}
def fetch_feed(url: str, timeout: int = 5) -> feedparser.FeedParserDict:
"""
Retrieve the RSS feed using ``requests`` so we can enforce a timeout.
Raises ConnectionError on network failure and MissingEntriesError if the feed is empty.
"""
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status()
except requests.RequestException as exc:
raise ConnectionError(f"Failed to fetch RSS feed: {exc}") from exc
feed = feedparser.parse(response.content)
if not feed.entries:
raise MissingEntriesError("The RSS feed contains no entries")
return feed
def dispatch_entries(feed: feedparser.FeedParserDict) -> List[dict]:
"""
Iterate over feed entries, find the first matching handler, and collect
the processed payloads. Unmatched entries are logged and ignored.
"""
processed = []
for entry in feed.entries:
matched = False
for pattern, func in HANDLERS.items():
if pattern.search(entry.title):
try:
payload = func(entry)
processed.append(payload)
print(f"[{payload['type']}] – {payload['title']}")
except Exception as exc:
print(f"Handler error for entry '{entry.title}': {exc}", file=sys.stderr)
matched = True
break
if not matched:
print(f"[UNMATCHED] – {entry.title}")
return processed
def save_to_json(data: List[dict], path: Path) -> None:
"""Write the categorized payloads to a JSON file, one entry per line."""
with path.open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def main() -> None:
RSS_URL = "https://discord.com/blog/rss.xml"
try:
feed = fetch_feed(RSS_URL, timeout=5)
except (ConnectionError, MissingEntriesError) as err:
print(f"Fatal error: {err}", file=sys.stderr)
sys.exit(1)
categorized = dispatch_entries(feed)
if not categorized:
print("No entries were categorized; exiting.", file=sys.stderr)
sys.exit(0)
output_path = Path("categorized_entries.json")
save_to_json(categorized, output_path)
print(f"✅ Saved {len(categorized)} entries to {output_path}")
if __name__ == "__main__":
main()