In today’s fast‑paced digital world, staying updated with the latest news, blog posts, and podcast episodes is essential for developers, marketers, and content creators alike. A powerful way to automate this process is by building a Python RSS feed parser and aggregator. Whether you want to monitor industry trends, curate content for a newsletter, or power a custom dashboard, Python offers robust libraries and straightforward techniques to fetch, parse, and combine RSS feeds into a single, searchable feed. In this guide, we’ll explore the fundamentals of RSS, walk through a step‑by‑step implementation using the popular feedparser library, and show you how to extend the solution with caching, filtering, and a simple web interface.
What Is an RSS Feed and Why Parse It?
RSS (Really Simple Syndication) is an XML‑based format that allows websites to publish a stream of content updates—such as articles, videos, or podcasts—in a machine‑readable way. By parsing RSS feeds, you can:
- Aggregate content from multiple sources into one place.
- Automate content discovery for SEO and content marketing.
- Build custom notifications, email digests, or mobile alerts.
- Analyze trends by extracting metadata like publish dates and categories.
Because RSS is standardized, a single parser can handle thousands of different sites, making it an ideal backbone for any content‑crawling project.
Choosing the Right Python Library
Several Python packages can read RSS, but feedparser stands out for its simplicity, active maintenance, and comprehensive support for both RSS 2.0 and Atom feeds. It abstracts the XML parsing, handling namespaces, date conversion, and character encoding automatically.
Alternative options include:
xml.etree.ElementTree– low‑level XML parsing, more control but more code.BeautifulSoup– great for messy HTML but overkill for well‑formed RSS.lxml– high performance, but requires extra dependencies.
For most use cases, feedparser provides the perfect balance of ease‑of‑use and power.
Setting Up the Development Environment
Install Required Packages
pip install feedparser requests beautifulsoup4
We include requests for reliable HTTP fetching and beautifulsoup4 for optional HTML sanitization of feed summaries.
Project Structure
aggregator.py– core parsing and aggregation logic.cache.py– simple file‑based caching to avoid redundant network calls.templates/– optional Jinja2 templates for a web UI.
Building a Basic RSS Parser
Fetching and Parsing a Single Feed
import feedparser
import requests
def fetch_feed(url):
"""
Retrieve an RSS/Atom feed and return a parsed FeedParserDict.
"""
response = requests.get(url, timeout=10)
response.raise_for_status()
return feedparser.parse(response.content)
# Example usage
feed_url = 'https://realpython.com/atom.xml'
feed = fetch_feed(feed_url)
print(f"Feed title: {feed.feed.title}")
for entry in feed.entries[:5]:
print(f"- {entry.title} ({entry.link})")
This snippet demonstrates the core of a parser: download the XML, let feedparser handle the heavy lifting, and access common fields like title, link, and published.
Normalizing Entries
Different publishers may use slightly different field names. To build a reliable aggregator, we normalize each entry into a consistent dictionary.
from datetime import datetime
from bs4 import BeautifulSoup
def normalize_entry(entry):
"""
Convert a raw feed entry into a uniform dict.
"""
# Helper to safely get attribute values
def get(attr, default=''):
return getattr(entry, attr, default)
# Parse the published date into a datetime object
published = get('published', get('updated'))
try:
published_dt = datetime(*entry.published_parsed[:6])
except Exception:
published_dt = None
# Clean HTML from summary/description
raw_summary = get('summary', get('description'))
clean_summary = BeautifulSoup(raw_summary, 'html.parser').get_text()
return {
'title': get('title'),
'link': get('link'),
'author': get('author', ''),
'published': published_dt,
'summary': clean_summary,
'tags': [tag.term for tag in getattr(entry, 'tags', [])],
'source': entry.get('source', {}).get('title', '')
}
The normalize_entry function ensures that downstream code can rely on a predictable schema, regardless of the feed’s original format.
Aggregating Multiple Feeds
Defining the Feed List
Store your target URLs in a simple Python list or a configuration file (JSON/YAML). For illustration, we’ll use a hard‑coded list.
FEED_URLS = [
'https://realpython.com/atom.xml',
'https://www.reddit.com/r/python/.rss',
'https://feeds.feedburner.com/TechCrunch/',
'https://xkcd.com/atom.xml'
]
Combining and Sorting Entries
def aggregate_feeds(urls):
all_entries = []
for url in urls:
try:
feed = fetch_feed(url)
for raw_entry in feed.entries:
entry = normalize_entry(raw_entry)
entry['source'] = feed.feed.get('title', url)
all_entries.append(entry)
except Exception as e:
print(f"Failed to process {url}: {e}")
# Sort by newest first
all_entries.sort(key=lambda e: e['published'] or datetime.min, reverse=True)
return all_entries
# Get the combined feed
combined = aggregate_feeds(FEED_URLS)
print(f"Collected {len(combined)} items.")
After aggregation, you have a single list of dictionaries that can be rendered as JSON, stored in a database, or displayed on a web page.
Adding Caching to Reduce Network Load
Repeatedly pulling the same feeds can quickly exhaust rate limits or waste bandwidth. A lightweight file‑based cache works well for most hobby projects.
import os
import json
import hashlib
import time
CACHE_DIR = 'feed_cache'
CACHE_TTL = 60 * 15 # 15 minutes
def cache_path(url):
filename = hashlib.sha256(url.encode('utf-8')).hexdigest() + '.json'
return os.path.join(CACHE_DIR, filename)
def load_from_cache(url):
path = cache_path(url)
if not os.path.exists(path):
return None
if time.time() - os.path.getmtime(path) > CACHE_TTL:
return None
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
def save_to_cache(url, data):
os.makedirs(CACHE_DIR, exist_ok=True)
path = cache_path(url)
with open(path, 'w', encoding='utf-8') as f:
json.dump(data, f)
def fetch_feed_cached(url):
cached = load_from_cache(url)
if cached:
return feedparser.FeedParserDict(cached)
response = requests.get(url, timeout=10)
response.raise_for_status()
parsed = feedparser.parse(response.content)
save_to_cache(url, parsed)
return parsed
Replace fetch_feed with fetch_feed_cached inside aggregate_feeds to enable caching automatically.
Filtering and Tagging Content
Most aggregators need the ability to filter out noise—spammy posts, irrelevant categories, or outdated items. Below is a simple filter that excludes entries older than a week and only keeps posts containing certain keywords.
from datetime import timedelta
KEYWORDS = ['Python', 'Django', 'Flask', 'Machine Learning']
MAX_AGE = timedelta(days=7)
def filter_entries(entries):
now = datetime.utcnow()
filtered = []
for e in entries:
if not e['published']:
continue
if now - e['published'] > MAX_AGE:
continue
if any(kw.lower() in e['title'].lower() for kw in KEYWORDS):
filtered.append(e)
return filtered
# Apply filter
relevant = filter_entries(combined)
print(f"{len(relevant)} items match our criteria.")
Custom filters can also be built around tags, author, or even the content of summary.
Serving the Aggregated Feed via a Simple Flask API
To make the aggregated data consumable by other applications, expose it through a lightweight web service.
from flask import Flask, jsonify
app = Flask(__name__)
@app.route('/api/aggregated')
def api_aggregated():
entries = aggregate_feeds(FEED_URLS)
entries = filter_entries(entries)
# Convert datetime objects to ISO strings for JSON serialization
for e in entries:
e['published'] = e['published'].isoformat() if e['published'] else None
return jsonify(entries)
if __name__ == '__main__':
app.run(debug=True, port=5000)
Running this script starts a local server at
Leave a Reply