Python Rss Feed Parser And Aggregator

In the age of information overload, staying up‑to‑date with the latest news, blog posts, or product updates can feel like a full‑time job. Fortunately, Python makes it easy to automate the collection and display of content through RSS feeds. In this guide you’ll learn how to build a robust Python RSS feed parser and aggregator from scratch, explore the most popular libraries, and discover best practices for handling XML, caching results, and presenting a clean output for your users.

What Is an RSS Feed and Why Parse It?

RSS (Really Simple Syndication) is an XML‑based format that allows publishers to broadcast headlines, summaries, and links to new content. By parsing these feeds, you can:

• Consolidate multiple sources into a single dashboard.
• Trigger automated workflows (e.g., email alerts, Slack notifications).
• Power custom news readers or content recommendation engines.

Because RSS is standardized, a well‑written parser can work with thousands of sites without needing site‑specific scraping logic.

Choosing the Right Python Library

Several Python packages simplify RSS parsing. The table below highlights the most common options and their ideal use‑cases.

Setting Up the Project

Start by creating a virtual environment and installing the required packages:

python -m venv rss_env
source rss_env/bin/activate  # On Windows use `rss_env\Scripts\activate`
pip install feedparser requests lxml beautifulsoup4

Basic Feed Parsing with feedparser

The feedparser library abstracts away most of the XML intricacies. Below is a minimal example that fetches and prints the latest five entries from a feed.

import feedparser

def fetch_latest(feed_url, limit=5):
    parsed = feedparser.parse(feed_url)
    if parsed.bozo:
        # bozo flag indicates a parsing error
        raise ValueError(f"Failed to parse feed: {parsed.bozo_exception}")

    entries = parsed.entries[:limit]
    for entry in entries:
        title = entry.get('title', 'No title')
        link = entry.get('link', '#')
        published = entry.get('published', 'No date')
        print(f"{title}\n{published}\n{link}\n{'-'*40}")

if __name__ == "__main__":
    RSS_URL = "https://news.ycombinator.com/rss"
    fetch_latest(RSS_URL)

This script demonstrates three essential steps:

1. Download and parse the XML with feedparser.parse().
2. Validate the result using the bozo flag.
3. Iterate over entries and extract common fields.

Advanced Parsing with requests + ElementTree

When you need to add custom HTTP headers, authentication, or cache control, combine requests with the standard library’s XML parser.

import requests
import xml.etree.ElementTree as ET

def fetch_feed(url):
    headers = {
        "User-Agent": "MyRSSAggregator/1.0 (+https://example.com)",
        "Accept": "application/rss+xml, application/xml"
    }
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()  # Raise on HTTP errors
    return response.content

def parse_feed(xml_data):
    root = ET.fromstring(xml_data)
    # RSS 2.0 uses  → 
    for item in root.findall("./channel/item"):
        title = item.findtext("title", default="No title")
        link = item.findtext("link", default="#")
        pub_date = item.findtext("pubDate", default="No date")
        print(f"{title}\n{pub_date}\n{link}\n{'='*30}")

if __name__ == "__main__":
    feed_xml = fetch_feed("https://xkcd.com/atom.xml")
    parse_feed(feed_xml)

Key advantages of this approach:

• Full control over request parameters (e.g., If-Modified-Since for conditional GET).
• Ability to handle namespaces or custom tags that feedparser might ignore.

Building a Simple RSS Aggregator

Now that we can parse individual feeds, let’s combine several sources into one unified view. The following example stores feed URLs in a list, fetches each asynchronously using aiohttp, and renders a plain‑text summary.

import asyncio
import aiohttp
import feedparser

FEED_URLS = [
    "https://news.ycombinator.com/rss",
    "https://www.reddit.com/r/python/.rss",
    "https://xkcd.com/atom.xml"
]

async def fetch(session, url):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.text()

async def parse_and_collect(url, content):
    parsed = feedparser.parse(content)
    items = []
    for entry in parsed.entries[:3]:  # Grab top 3 items per feed
        items.append({
            "title": entry.get("title", "No title"),
            "link": entry.get("link", "#"),
            "source": parsed.feed.get("title", "Unknown"),
            "published": entry.get("published", "No date")
        })
    return items

async def aggregate():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in FEED_URLS]
        raw_contents = await asyncio.gather(*tasks, return_exceptions=True)

        all_items = []
        for url, content in zip(FEED_URLS, raw_contents):
            if isinstance(content, Exception):
                print(f"Error fetching {url}: {content}")
                continue
            items = await parse_and_collect(url, content)
            all_items.extend(items)

        # Sort by published date (most recent first)
        all_items.sort(key=lambda x: x["published"], reverse=True)
        for i, item in enumerate(all_items, 1):
            print(f"{i}. [{item['source']}] {item['title']}")
            print(f"   {item['published']}")
            print(f"   {item['link']}\n")

if __name__ == "__main__":
    asyncio.run(aggregate())

Highlights of this aggregator:

• Asynchronous I/O speeds up fetching dozens of feeds.
• Results are normalized into a common dictionary structure.
• Sorting by published creates a chronological “news wall”.

Best Practices for a Production‑Ready Aggregator

While the examples above work well for learning, a real‑world service needs extra layers of reliability, performance, and security.

1. Caching and Conditional Requests

Repeatedly downloading the same XML wastes bandwidth and can get your IP blocked. Use HTTP caching headers (ETag, Last-Modified) to request only changed content.

headers = {"If-None-Match": cached_etag, "If-Modified-Since": cached_date}
response = requests.get(url, headers=headers)
if response.status_code == 304:
    # No changes – use the cached version
    xml = cached_xml
else:
    xml = response.text
    # Store new ETag/Last‑Modified for next call

2. Error Handling and Resilience

Feeds can be malformed, temporarily unavailable, or return unexpected status codes. Wrap network calls in try/except blocks, log errors with context, and implement exponential back‑off for retries.

import time
import logging

def safe_fetch(url, retries=3, backoff=2):
    for attempt in range(retries):
        try:
            return fetch_feed(url)
        except Exception as e:
            logging.warning(f"Attempt {attempt+1} failed for {url}: {e}")
            time.sleep(backoff ** attempt)
    raise RuntimeError(f"All attempts failed for {url}")

3. Normalizing Dates

RSS and Atom use various date formats (RFC 822, ISO 8601). Convert them to datetime objects with dateutil.parser for reliable sorting.

from dateutil import parser as dt_parser

def to_datetime(date_str):
    try:
        return dt_parser.parse(date_str)
    except Exception:
        return None

4. Storing Aggregated Data

Depending on your use‑case, you might