Key Takeaways
- Utilize `functools.wraps` and `functools.partial` to build configurable, parametrized decorators that preserve function metadata and allow flexible argument passing.
- Implement stateful decorators using closures and local SQLite databases to maintain persistent idempotency caches across executions, preventing duplicate data ingestion.
- Dynamically intercept network and parsing exceptions within decorators to apply sophisticated exponential backoff and retry strategies without altering core business logic.
- Enforce strict schema validation and type coercion on incoming data payloads at the decorator boundary, ensuring data quality before downstream processing.
- Compose multiple decorators seamlessly to construct robust, layered data ingestion pipelines that remain clean, testable, and maintainable.
The Problem
Our goal is to reliably ingest new posts from the Cloudflare Engineering Blog RSS feed. In a production environment, this isn't just about fetching XML and parsing it. We face several challenges: 1. **Network Instability:** The internet is flaky. We need to handle temporary connection drops or timeouts gracefully with retries. 2. **Idempotency:** RSS feeds can sometimes serve the same entries multiple times, or our ingestion job might restart and re-process recent entries. We absolutely cannot have duplicate records in our data lake. 3. **Data Quality:** RSS feeds, while structured, can have missing fields, unexpected types, or malformed content. We need to validate each entry before it enters our system. 4. **Code Bloat:** Mixing these concerns directly into the function that fetches and processes the feed makes the code hard to read, test, and maintain.Data and Sources
We'll be using the live RSS feed from the Cloudflare Engineering Blog.- Cloudflare Engineering Blog RSS endpoint: https://blog.cloudflare.com/rss/
- Feedparser library documentation: https://pythonhosted.org/feedparser/
- Official Python `functools` documentation: https://docs.python.org/3/library/functools.html
- Python `sqlite3` documentation: https://docs.python.org/3/library/sqlite3.html
Step 1 — Constructing the Stateful Deduplication Closure
The first challenge is idempotency. We need a way to remember which RSS entries we've already processed, even across different runs of our ingestion script. A simple in-memory set won't suffice for persistence. For local, single-instance deployments, SQLite is a perfect lightweight solution. We'll create a decorator that uses a SQLite database to store a hash of each processed entry's unique identifier (like its `link` or `id`). This decorator will initialize a SQLite connection, create a `seen_entries` table if it doesn't exist, and then, for each entry, check if its hash is already in the table. If it is, the entry is skipped. If not, the entry is processed by the wrapped function, and its hash is then stored.import sqlite3
import hashlib
import functools
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def idempotent(db_path="idempotency_cache.db"):
def decorator(func):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS seen_entries (hash TEXT PRIMARY KEY)")
conn.commit()
@functools.wraps(func)
def wrapper(entries, *args, **kwargs):
processed_entries = []
for entry in entries:
entry_id = entry.get('link', entry.get('id', str(entry)))
entry_hash = hashlib.sha256(entry_id.encode('utf-8')).hexdigest()
cursor.execute("SELECT 1 FROM seen_entries WHERE hash = ?", (entry_hash,))
if cursor.fetchone():
logging.info(f"Skipping duplicate entry: {entry_id}")
continue
# If not seen, process it and then mark as seen
processed_entry = func([entry], *args, **kwargs) # Pass as list for consistency
if processed_entry:
processed_entries.extend(processed_entry)
cursor.execute("INSERT INTO seen_entries (hash) VALUES (?)", (entry_hash,))
conn.commit()
return processed_entries
return wrapper
return decorator
Step 2 — Parametrizing Decorators with Exception Interception
Next, we need resilience against network failures. An exponential backoff retry mechanism is standard practice. This decorator will catch specific exceptions (like `URLError` from network issues or `feedparser`'s own parsing errors) and retry the wrapped function call with increasing delays.import time
import random
from urllib.error import URLError
import feedparser # We'll catch feedparser's internal errors too
def with_backoff(retries=3, initial_delay=1, max_delay=60, errors=(URLError, feedparser.FeedParserDict)): # FeedParserDict for potential parsing errors
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for i in range(retries):
try:
return func(*args, **kwargs)
except errors as e:
logging.warning(f"Attempt {i+1}/{retries} failed with {type(e).__name__}: {e}")
if i < retries - 1:
sleep_time = min(delay, max_delay) + random.uniform(0, 0.5 * delay) # Add jitter
logging.info(f"Retrying in {sleep_time:.2f} seconds...")
time.sleep(sleep_time)
delay *= 2
else:
logging.error(f"All {retries} attempts failed. Giving up.")
raise # Re-raise the last exception if all retries fail
return None # Should ideally not be reached if exception is always re-raised
return wrapper
return decorator
Step 3 — Schema Enforcement and Payload Reshaping
Raw RSS entries are dictionaries, and their structure can vary. We want to ensure that only entries conforming to a strict schema are processed and that they are converted into a consistent format. Python `dataclasses` are perfect for defining this schema. Our decorator will attempt to convert each raw entry dictionary into our defined `dataclass` instance, dropping or logging entries that don't match.from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import datetime
@dataclass
class CloudflareEntry:
title: str
link: str
published: datetime.datetime
summary: Optional[str] = None
authors: List[str] = field(default_factory=list)
tags: List[str] = field(default_factory=list)
# Store original raw data for debugging if needed
_raw_data: Dict[str, Any] = field(repr=False, default_factory=dict)
def validate_entry(schema_cls):
def decorator(func):
@functools.wraps(func)
def wrapper(entries: List[Dict[str, Any]], *args, **kwargs) -> List[Any]:
validated_entries = []
for raw_entry in entries:
try:
# Basic mapping from feedparser keys to dataclass fields
# This could be more sophisticated with a dedicated mapping library
processed_entry = {
'title': raw_entry.get('title'),
'link': raw_entry.get('link'),
'published': datetime.datetime.fromtimestamp(
time.mktime(raw_entry.get('published_parsed'))
) if raw_entry.get('published_parsed') else None,
'summary': raw_entry.get('summary'),
'authors': [author.get('name') for author in raw_entry.get('authors', []) if author.get('name')],
'tags': [tag.get('term') for tag in raw_entry.get('tags', []) if tag.get('term')],
'_raw_data': raw_entry # Keep raw data for context
}
# Filter out entries with missing critical fields
if not all([processed_entry['title'], processed_entry['link'], processed_entry['published']]):
raise ValueError("Missing critical fields (title, link, published)")
validated_entry = schema_cls(**processed_entry)
validated_entries.append(validated_entry)
except (TypeError, ValueError, AttributeError) as e:
logging.warning(f"Skipping malformed entry due to validation error: {e}. Raw data: {raw_entry.get('link', 'N/A')}")
return func(validated_entries, *args, **kwargs) # Pass validated dataclass instances
return wrapper
return decorator
Step 4 — Composing Multiple Decorators for Production Pipelines
Now we bring it all together. Python's decorator syntax allows stacking multiple decorators, which are applied from the bottom up. This means the innermost decorator is applied first to the function, then the next one wraps that result, and so on. Let's define our core RSS fetching and processing logic.import feedparser
import pandas as pd
import os
# Define the core RSS fetcher function
@with_backoff(retries=5, initial_delay=2, max_delay=120)
@idempotent(db_path="cloudflare_rss_cache.db")
@validate_entry(schema_cls=CloudflareEntry)
def fetch_and_process_cloudflare_rss(url: str, entries_to_process: Optional[int] = None) -> List[CloudflareEntry]:
"""
Fetches, processes, validates, and deduplicates entries from the Cloudflare RSS feed.
"""
logging.info(f"Attempting to fetch RSS feed from {url}")
feed = feedparser.parse(url)
if feed.bozo:
logging.error(f"Feed parsing error (bozo bit set): {feed.bozo_exception}")
# Depending on severity, you might want to raise or return []
# For now, we'll try to process what's available
# The entries are passed through the decorators in reverse order of application
# @validate_entry will receive raw feed.entries
# @idempotent will receive validated CloudflareEntry objects from @validate_entry
# @with_backoff wraps the whole thing, handling network errors during feedparser.parse
# We yield the entries to the next decorator in the stack.
# The actual processing logic here is minimal, as decorators handle much of it.
# The @validate_entry decorator will call this function with already validated entries.
# The @idempotent decorator will call this function with new, validated entries.
# This specific function signature expects a list of entries and returns a list.
# When called by @idempotent, it's called with a single-item list.
# The actual feedparser.parse(url) is implicitly part of the call that hits @with_backoff.
# Let's refactor slightly to make the core function cleaner for the decorator stack.
# The `fetch_cloudflare_feed` will be the one directly decorated.
# The `idempotent` and `validate_entry` decorators will operate on the list of entries *returned* by `fetch_cloudflare_feed`.
# This means the `idempotent` and `validate_entry` decorators need to be designed to wrap a function that takes a list of entries.
# Let's adjust the decorator application slightly for clarity:
# @with_backoff will wrap the network call.
# @idempotent and @validate_entry will wrap a *processing* function.
# Re-evaluating the decorator flow:
# 1. @with_backoff wraps the actual `feedparser.parse` call.
# 2. The *result* of that call (the list of entries) then goes through @idempotent.
# 3. The *result* of @idempotent then goes through @validate_entry.
# This implies the core function should return raw entries, and subsequent decorators process them.
# Let's define a helper to get raw entries first, then process.
# This will make the decorator chain cleaner.
all_entries = []
if hasattr(feed, 'entries'):
all_entries = feed.entries
if entries_to_process:
all_entries = all_entries[:entries_to_process]
# The actual work of processing (validation, deduplication) happens *after* fetching.
# The current decorator design for idempotent and validate_entry expects a list of entries.
# So the decorated function itself should be the *processor* of entries.
# Let's adjust the definition of `fetch_and_process_cloudflare_rss`
# It will be called *by* the decorators, receiving a single entry (from idempotent)
# that has already passed validation.
# This function will now simply "yield" or return the processed entry
# after it has been through the validation and idempotency checks.
# The @idempotent decorator is designed to call `func` with a *list* of [entry].
# So `func` should expect `List[Dict]` or `List[CloudflareEntry]`.
# Let's revise the `fetch_and_process_cloudflare_rss` to be the actual `_process_single_entry`
# and have a main loop that calls the decorated function.
# This is where understanding decorator composition becomes key.
# The `fetch_and_process_cloudflare_rss` function itself needs to return a list of *raw* entries
# for the @idempotent and @validate_entry decorators to work on.
# Let's create two functions: one for fetching, one for processing.
# The fetching one will be wrapped by @with_backoff.
# The processing one will be wrapped by @idempotent and @validate_entry.
pass # This will be defined in the complete script.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import sqlite3
import hashlib
import functools
import logging
import time
import random
from urllib.error import URLError
import feedparser
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
import datetime
import pandas as pd
import os
# --- Configuration ---
RSS_FEED_URL = 'https://blog.cloudflare.com/rss/'
DB_PATH = "cloudflare_rss_cache.db"
OUTPUT_FILE = "cloudflare_processed_entries.parquet"
LOG_LEVEL = logging.INFO
# --- Logging Setup ---
logging.basicConfig(level=LOG_LEVEL, format='%(asctime)s - %(levelname)s - %(message)s')
# --- Dataclass for Schema Enforcement ---
@dataclass
class CloudflareEntry:
title: str
link: str
published: datetime.datetime
summary: Optional[str] = None
authors: List[str] = field(default_factory=list)
tags: List[str] = field(default_factory=list)
_raw_data: Dict[str, Any] = field(repr=False, default_factory=dict) # For debugging/auditing
def to_dict(self):
return {
"title": self.title,
"link": self.link,
"published": self.published.isoformat(),
"summary": self.summary,
"authors": self.authors,
"tags": self.tags,
}
# --- Decorator 1: Idempotency with SQLite ---
def idempotent(db_path: str = "idempotency_cache.db"):
def decorator(func):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS seen_entries (hash TEXT PRIMARY KEY)")
conn.commit()
@functools.wraps(func)
def wrapper(entries: List[Dict[str, Any]], *args, **kwargs) -> List[Any]:
processed_results = []
for entry in entries:
# Use link as primary identifier, fall back to id, then string representation
entry_id = entry.get('link', entry.get('id', str(entry)))
entry_hash = hashlib.sha256(entry_id.encode('utf-8')).hexdigest()
cursor.execute("SELECT 1 FROM seen_entries WHERE hash = ?", (entry_hash,))
if cursor.fetchone():
logging.info(f"Skipping duplicate entry (hash: {entry_hash[:8]}...): {entry_id}")
continue
# Process the entry with the wrapped function
# The wrapped function (`_process_single_entry`) expects a single entry
result = func([entry], *args, **kwargs) # Pass as list for consistency with how validate_entry expects it
if result: # If func returns a processed entry (e.g., CloudflareEntry)
processed_results.extend(result)
cursor.execute("INSERT INTO seen_entries (hash) VALUES (?)", (entry_hash,))
conn.commit()
return processed_results
return wrapper
return decorator
# --- Decorator 2: Exponential Backoff for Network Calls ---
def with_backoff(retries: int = 3, initial_delay: int = 1, max_delay: int = 60, errors=(URLError,)):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for i in range(retries):
try:
return func(*args, **kwargs)
except errors as e:
logging.warning(f"Attempt {i+1}/{retries} failed with {type(e).__name__}: {e}")
if i < retries - 1:
sleep_time = min(delay, max_delay) + random.uniform(0, 0.5 * delay) # Add jitter
logging.info(f"Retrying in {sleep_time:.2f} seconds...")
time.sleep(sleep_time)
delay *= 2
else:
logging.error(f"All {retries} attempts failed. Giving up.")
raise # Re-raise the last exception if all retries fail
return None # Should not be reached
return wrapper
return decorator
# --- Decorator 3: Schema Validation and Reshaping ---
def validate_entry(schema_cls: type):
def decorator(func):
@functools.wraps(func)
def wrapper(entries: List[Dict[str, Any]], *args, **kwargs) -> List[Any]:
validated_instances = []
for raw_entry in entries:
try:
# Map feedparser entry to CloudflareEntry dataclass
processed_entry_data = {
'title': raw_entry.get('title'),
'link': raw_entry.get('link'),
'published': datetime.datetime.fromtimestamp(
time.mktime(raw_entry.get('published_parsed'))
) if raw_entry.get('published_parsed') else None,
'summary': raw_entry.get('summary'),
'authors': [author.get('name') for author in raw_entry.get('authors', []) if author.get('name')],
'tags': [tag.get('term') for tag in raw_entry.get('tags', []) if tag.get('term')],
'_raw_data': raw_entry
}
# Basic validation for critical fields
if not all([processed_entry_data['title'], processed_entry_data['link'], processed_entry_data['published']]):
raise ValueError("Missing critical fields (title, link, published)")
validated_instance = schema_cls(**processed_entry_data)
validated_instances.append(validated_instance)
except (TypeError, ValueError, AttributeError, KeyError) as e:
logging.warning(f"Skipping malformed entry due to validation error: {e}. Link: {raw_entry.get('link', 'N/A')}")
# Pass only the successfully validated instances to the wrapped function
return func(validated_instances, *args, **kwargs)
return wrapper
return decorator
# --- Core Functions ---
@with_backoff(retries=5, initial_delay=2, max_delay=120, errors=(URLError, feedparser.FeedParserDict))
def _fetch_raw_feed_entries(url: str) -> List[Dict[str, Any]]:
"""Fetches the raw RSS feed and returns a list of its entries."""
logging.info(f"Fetching RSS feed from {url}")
try:
feed = feedparser.parse(url)
if feed.bozo:
logging.warning(f"Feed parsing warnings/errors (bozo bit set): {feed.bozo_exception}")
# Depending on strictness, we might raise here or proceed with partial data.
# For now, we proceed.
return feed.entries
except Exception as e: # Catch all for feedparser's internal issues not covered by URLError
logging.error(f"Failed to parse RSS feed {url}: {e}")
raise # Re-raise to be caught by @with_backoff if it's in `errors` tuple, otherwise fail.
@idempotent(db_path=DB_PATH)
@validate_entry(schema_cls=CloudflareEntry)
def _process_single_entry(entries: List[CloudflareEntry]) -> List[CloudflareEntry]:
"""
This function acts as the final processing step for a single,
already validated and deduplicated CloudflareEntry.
It receives a list containing one CloudflareEntry object.
"""
# At this point, `entries` will contain a list of `CloudflareEntry` instances,
# already validated and confirmed as not duplicated by the decorators above.
# We simply return them to be collected by the outer loop.
return entries
# --- Main Execution Logic ---
def main():
logging.info("Starting Cloudflare RSS ingestion pipeline...")
# 1. Fetch raw entries (with retries)
try:
raw_entries = _fetch_raw_feed_entries(RSS_FEED_URL)
logging.info(f"Successfully fetched {len(raw_entries)} raw entries.")
except Exception as e:
logging.error(f"Critical error fetching RSS feed: {e}. Exiting.")
return
# 2. Process entries through idempotency and validation decorators
processed_cloudflare_entries: List[CloudflareEntry] = []
# The _process_single_entry function is designed to be called by the @idempotent decorator.
# @idempotent iterates over raw_entries, then calls `_process_single_entry` with `[raw_entry]`
# for each *new* raw_entry.
# The @validate_entry decorator then transforms `[raw_entry]` into `[CloudflareEntry]`
# before calling the actual `_process_single_entry` function.
# We need to explicitly call the decorated _process_single_entry with the list of raw entries
# so that the @idempotent decorator can do its work.
# The `idempotent` decorator expects a list of raw dictionaries as its first argument
# and then passes a single item list `[entry]` to the decorated `func`.
# So `_process_single_entry` needs to be able to handle a list of raw_entries as its *first* call,
# and then subsequent calls from the `idempotent` decorator will be with `[raw_entry]`.
# This implies a slight adjustment to the `idempotent` and `validate_entry` decorators
# to receive the *full list* and then process them internally.
# Or, the main loop explicitly calls the decorated processor *per raw entry*.
# Let's adjust the `idempotent` and `validate_entry` to work on a list passed ONCE.
# This is a more common pattern for batch-like processing.
# The `idempotent` decorator will filter, `validate_entry` will validate.
# The `_process_single_entry` will just be the final sink.
# REVISED DECORATOR APPLICATION:
# `process_feed_data` will be the function that takes all fetched raw entries.
# It will be decorated by @idempotent and @validate_entry.
@idempotent(db_path=DB_PATH)
@validate_entry(schema_cls=CloudflareEntry)
def process_feed_data(entries: List[Dict[str, Any]]) -> List[CloudflareEntry]:
"""
Receives a list of raw feed entries.
Decorators handle deduplication and validation.
This function then simply returns the already-processed CloudflareEntry objects.
"""
# At this point, `entries` is already a list of `CloudflareEntry` instances,
# having passed through @idempotent (which filtered duplicates) and
# @validate_entry (which converted and validated).
logging.info(f"Core processing function received {len(entries)} new, validated entries.")
return entries # The decorators have done the heavy lifting
try:
processed_cloudflare_entries = process_feed_data(raw_entries)
logging.info(f"Finished processing. {len(processed_cloudflare_entries)} new entries ingested.")
except Exception as e:
logging.error(f"Error during entry processing: {e}. Some entries might have been skipped.")
# We might still have partially processed data in processed_cloudflare_entries
# depending on where the error occurred in the decorator chain.
if not processed_cloudflare_entries:
logging.info("No new entries to save.")
return
# 3. Save processed entries to Parquet
try:
df = pd.DataFrame([entry.to_dict() for entry in processed_cloudflare_entries])
# Append to existing file or create new
if os.path.exists(OUTPUT_FILE):
existing_df = pd.read_parquet(OUTPUT_FILE)
combined_df = pd.concat([existing_df, df]).drop_duplicates(subset=['link']).reset_index(drop=True)
combined_df.to_parquet(OUTPUT_FILE, index=False)
logging.info(f"Appended {len(df)} new entries to {OUTPUT_FILE}. Total unique entries: {len(combined_df)}")
else:
df.to_parquet(OUTPUT_FILE, index=False)
logging.info(f"Saved {len(df)} new entries to {OUTPUT_FILE}.")
except Exception as e:
logging.error(f"Error saving data to Parquet file {OUTPUT_FILE}: {e}")
if __name__ == "__main__":
main()
Expected Output
When you run the script (`python your_script_name.py`), you should see console output similar to this:2024-07-29 10:30:01,123 - INFO - Starting Cloudflare RSS ingestion pipeline...
2024-07-29 10:30:01,123 - INFO - Fetching RSS feed from https://blog.cloudflare.com/rss/
2024-07-29 10:30:03,456 - INFO - Successfully fetched 20 raw entries.
2024-07-29 10:30:03,457 - INFO - Core processing function received 5 new, validated entries.
2024-07-29 10:30:03,458 - INFO - Finished processing. 5 new entries ingested.
2024-07-29 10:30:03,459 - INFO - Saved 5 new entries to cloudflare_processed_entries.parquet.
2024-07-29 10:30:05,123 - INFO - Starting Cloudflare RSS ingestion pipeline...
2024-07-29 10:30:05,123 - INFO - Fetching RSS feed from https://blog.cloudflare.com/rss/
2024-07-29 10:30:07,456 - INFO - Successfully fetched 20 raw entries.
2024-07-29 10:30:07,457 - INFO - Skipping duplicate entry (hash: f1a2b3c4...): https://blog.cloudflare.com/filter-reshape-redact-logs-transformers/
2024-07-29 10:30:07,458 - INFO - Skipping duplicate entry (hash: d5e6f7a8...): https://blog.cloudflare.com/introducing-clef-omni/
... (many more skipping messages)
2024-07-29 10:30:07,460 - INFO - Core processing function received 0 new, validated entries.
2024-07-29 10:30:07,461 - INFO - Finished processing. 0 new entries ingested.
2024-07-29 10:30:07,462 - INFO - No new entries to save.