How to Build a News Aggregator with Python

Building a news aggregator is one of the most practical Python projects you can undertake. Whether you want to track industry news for your team, monitor specific topics for research, or build a consumer-facing news product, the fundamentals are the same: connect to a reliable data source, filter the content you need, cache it locally, and present it to your users.

In this tutorial, you will build a fully functional news aggregator in Python from scratch. By the end, you will have a command-line tool that fetches and filters articles, a local SQLite cache to avoid redundant API calls, and an optional Flask web frontend to display your news feed in a browser. All of the code is complete and ready to run.

We will use the NewsMesh API as our data source. NewsMesh aggregates articles from over 6,000 news sources worldwide, enriches them with ML-powered categorization, topic extraction, people detection, and country relevance tagging. This gives us structured, queryable data rather than raw RSS feeds, which makes building a useful aggregator significantly easier. If you are new to news APIs in general, our guide on what a news API is and how it works covers the fundamentals.

Prerequisites

Before starting, make sure you have the following:

• Python 3.8 or higher installed on your machine. You can check with python3 --version.
• A NewsMesh API key. Sign up for free to get a development API key. The free tier includes access to all endpoints with a 24-hour delay on articles, which is perfect for building and testing.
• Basic familiarity with Python, HTTP requests, and working in a terminal.

Project Setup

Create a new project directory and set up a virtual environment. Using a virtual environment keeps your project dependencies isolated from your system Python.

mkdir news-aggregator
cd news-aggregator
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

Install the dependencies we will use throughout this tutorial:

pip install requests flask

Here is the project structure we will build toward:

news-aggregator/
    venv/
    newsclient.py      # API client class
    cache.py           # SQLite caching layer
    cli.py             # Command-line interface
    app.py             # Flask web frontend
    templates/
        index.html     # Article display template

Each file has a single responsibility, making the codebase easy to extend later. This separation matters because as your aggregator grows, you will want to swap components independently. For instance, you might replace the SQLite cache with Redis, or swap the CLI for a scheduled background job, without touching the API client code.

Step 1: Connect to the News API

The foundation of any news aggregator is a reliable connection to a data source. We will build a client class that wraps the NewsMesh API and provides clean Python methods for fetching articles.

The NewsMesh API uses straightforward REST conventions. You authenticate by passing your API key as a query parameter, and every response returns a JSON object with a data array of articles and a next_cursor field for pagination.

Create newsclient.py:

import requests
from typing import Optional


class NewsMeshClient:
    """Client for the NewsMesh News API."""

    BASE_URL = "https://api.newsmesh.co/v1"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.params = {"apiKey": self.api_key}

    def _get(self, endpoint: str, params: Optional[dict] = None) -> dict:
        """Make a GET request to the API and return the JSON response."""
        url = f"{self.BASE_URL}/{endpoint}"
        response = self.session.get(url, params=params or {})
        response.raise_for_status()
        return response.json()

    def get_trending(self, limit: int = 25) -> dict:
        """Fetch trending articles.

        Returns the most-read articles across all sources,
        ranked by engagement.
        """
        return self._get("trending", {"limit": limit})

    def get_latest(self, limit: int = 25, **kwargs) -> dict:
        """Fetch the latest articles with optional filters.

        Keyword arguments:
            category: Filter by category (e.g., "technology", "politics")
            country: ISO 3166-1 alpha-2 code (e.g., "us", "gb")
            from_date: Start date in YYYY-MM-DD format
            to_date: End date in YYYY-MM-DD format
            cursor: Pagination cursor from a previous response
        """
        params = {"limit": limit}
        if "category" in kwargs:
            params["category"] = kwargs["category"]
        if "country" in kwargs:
            params["country"] = kwargs["country"]
        if "from_date" in kwargs:
            params["from"] = kwargs["from_date"]
        if "to_date" in kwargs:
            params["to"] = kwargs["to_date"]
        if "cursor" in kwargs:
            params["cursor"] = kwargs["cursor"]
        return self._get("latest", params)

    def search(self, query: str, limit: int = 25, **kwargs) -> dict:
        """Search articles by keyword.

        The query supports advanced syntax:
            - "exact phrase" for phrase matching
            - +term for required terms
            - -term for excluded terms

        Keyword arguments:
            category: Filter by category
            country: ISO 3166-1 alpha-2 code
            from_date: Start date (YYYY-MM-DD)
            to_date: End date (YYYY-MM-DD)
            sort_by: "date_descending", "date_ascending", or "relevant"
            cursor: Pagination cursor
        """
        params = {"q": query, "limit": limit}
        if "category" in kwargs:
            params["category"] = kwargs["category"]
        if "country" in kwargs:
            params["country"] = kwargs["country"]
        if "from_date" in kwargs:
            params["from"] = kwargs["from_date"]
        if "to_date" in kwargs:
            params["to"] = kwargs["to_date"]
        if "sort_by" in kwargs:
            params["sortBy"] = kwargs["sort_by"]
        if "cursor" in kwargs:
            params["cursor"] = kwargs["cursor"]
        return self._get("search", params)

    def get_article(self, article_id: str) -> dict:
        """Fetch a single article by its ID."""
        return self._get(f"article/{article_id}")

Let us test the client. Open a Python shell and try fetching some articles:

from newsclient import NewsMeshClient

client = NewsMeshClient("your-api-key-here")

# Fetch trending articles
result = client.get_trending(limit=5)
for article in result["data"]:
    print(f"[{article['source']}] {article['title']}")

You should see output like:

[Reuters] Federal Reserve Holds Interest Rates Steady Amid Inflation Concerns
[BBC News] EU Reaches Agreement on New AI Regulation Framework
[The Guardian] Scientists Discover New Deep-Sea Species in Pacific Trench
[CNN] Tech Giants Report Record Quarterly Earnings
[Al Jazeera] Ceasefire Talks Resume in Regional Conflict

Each article object in the response contains these fields: article_id, title, description, link, published_date, source, category, topics, people, author, and media_url. The topics field is an array of specific subjects the article covers (like "Federal Reserve" or "interest rates"), while people lists individuals mentioned by name. This structured data is what makes building a useful aggregator possible without needing to do any natural language processing yourself. You get machine-readable metadata for every article, ready for filtering, grouping, and display.

Understanding Pagination

When you request articles, the API returns up to your specified limit per page. If more articles are available, the response includes a next_cursor value. Pass this cursor in your next request to get the following page:

# Fetch the first page
page1 = client.get_latest(limit=10, category="technology")

# If there are more results, fetch the next page
if page1.get("next_cursor"):
    page2 = client.get_latest(limit=10, category="technology",
                               cursor=page1["next_cursor"])

This cursor-based pagination is more reliable than offset-based pagination because new articles arriving between requests do not cause you to skip or duplicate items.

Step 2: Add Filtering

Raw news feeds are overwhelming. Thousands of articles are published every hour across global news sources, and most of them will be irrelevant to any given user. The core value of a news aggregator is not fetching articles -- it is filtering them. A well-designed filter system turns a firehose of information into a focused, useful feed. NewsMesh supports several filtering parameters that you can combine to narrow results precisely.

Filter by Category

NewsMesh uses ML-powered categorization to assign every article to a category. Available categories include: politics, technology, business, science, health, sports, entertainment, environment, and world. You can also pass multiple categories as a comma-separated string.

# Single category
tech_news = client.get_latest(limit=10, category="technology")

# Multiple categories
business_and_tech = client.get_latest(limit=10, category="technology,business")

Filter by Country

Country filtering uses ISO 3166-1 alpha-2 codes (two-letter codes like "us" for the United States, "gb" for the United Kingdom, "de" for Germany). NewsMesh offers two distinct country filters, and understanding the difference between them is important. The country parameter filters by article content relevance -- it returns articles that discuss events in the specified country, regardless of where the news outlet is based. The sourceCountry parameter, on the other hand, filters by the geographic origin of the news outlet itself. This distinction lets you answer different questions: "What is happening in Germany?" versus "What are German media outlets reporting about?"

# Articles about events in the United Kingdom
uk_news = client.get_latest(limit=10, country="gb")

# Articles about the US from UK-based outlets
us_from_uk = client.get_latest(limit=10, country="us")

Filter by Date Range

Use from_date and to_date parameters (YYYY-MM-DD format) to request articles from a specific time window. This is useful for historical research or building recap digests.

# Articles from a specific week
weekly_digest = client.get_latest(
    limit=25,
    category="technology",
    from_date="2026-01-20",
    to_date="2026-01-27"
)

Keyword Search

For more specific queries, the search endpoint supports full-text search with advanced syntax. You can use quoted phrases, required terms with +, excluded terms with -, and boolean operators.

# Search for a specific topic
ai_articles = client.search("artificial intelligence", limit=10)

# Require one term, exclude another
results = client.search('+renewable -nuclear energy policy', limit=10)

# Exact phrase match
results = client.search('"climate summit" 2026', limit=10,
                         sort_by="relevant")

Combining Filters

The real power comes from combining these filters. Here is a function that builds a targeted news feed:

def get_focused_feed(client, categories=None, countries=None,
                     keywords=None, days_back=7, limit=25):
    """Build a focused news feed with multiple filters."""
    from datetime import datetime, timedelta

    from_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")

    if keywords:
        return client.search(
            keywords,
            limit=limit,
            category=categories,
            country=countries,
            from_date=from_date,
            sort_by="relevant"
        )
    else:
        kwargs = {"limit": limit, "from_date": from_date}
        if categories:
            kwargs["category"] = categories
        if countries:
            kwargs["country"] = countries
        return client.get_latest(**kwargs)


# Example: Tech news about the US from the past 3 days
feed = get_focused_feed(
    client,
    categories="technology",
    countries="us",
    days_back=3
)

For a complete reference of all available endpoints and parameters, see the API documentation.

Step 3: Build a Simple Cache

Calling the API on every page load wastes your request quota and slows down your application. If five users visit your aggregator within the same minute and all request the same technology feed, that is five identical API calls returning nearly identical data. A local cache solves both problems: it reduces your API usage and makes repeated requests return instantly.

We will use SQLite because it requires no server setup, handles concurrent reads well, and is included in Python's standard library. The cache has two layers: a query cache that stores full API responses keyed by the request parameters, and an article store that keeps individual articles for offline querying.

Create cache.py:

import json
import sqlite3
import time
from typing import Optional


class ArticleCache:
    """SQLite-backed cache for news articles."""

    def __init__(self, db_path: str = "articles_cache.db"):
        self.conn = sqlite3.connect(db_path)
        self.conn.row_factory = sqlite3.Row
        self._create_tables()

    def _create_tables(self):
        """Create the cache tables if they do not exist."""
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS articles (
                article_id TEXT PRIMARY KEY,
                title TEXT NOT NULL,
                description TEXT,
                link TEXT,
                source TEXT,
                category TEXT,
                topics TEXT,
                people TEXT,
                author TEXT,
                media_url TEXT,
                published_date TEXT,
                cached_at REAL NOT NULL
            );

            CREATE TABLE IF NOT EXISTS query_cache (
                cache_key TEXT PRIMARY KEY,
                response_json TEXT NOT NULL,
                cached_at REAL NOT NULL
            );

            CREATE INDEX IF NOT EXISTS idx_articles_category
                ON articles(category);
            CREATE INDEX IF NOT EXISTS idx_articles_published
                ON articles(published_date DESC);
        """)
        self.conn.commit()

    def cache_articles(self, articles: list):
        """Store articles in the local cache."""
        now = time.time()
        for article in articles:
            self.conn.execute("""
                INSERT OR REPLACE INTO articles
                (article_id, title, description, link, source, category,
                 topics, people, author, media_url, published_date, cached_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                article["article_id"],
                article["title"],
                article.get("description"),
                article.get("link"),
                article.get("source"),
                article.get("category"),
                json.dumps(article.get("topics", [])),
                json.dumps(article.get("people")) if article.get("people") else None,
                article.get("author"),
                article.get("media_url"),
                article.get("published_date"),
                now,
            ))
        self.conn.commit()

    def cache_query_response(self, cache_key: str, response: dict,
                              ttl: int = 300):
        """Cache a full API response for a given query key."""
        self.conn.execute("""
            INSERT OR REPLACE INTO query_cache
            (cache_key, response_json, cached_at)
            VALUES (?, ?, ?)
        """, (cache_key, json.dumps(response), time.time()))
        self.conn.commit()

    def get_cached_response(self, cache_key: str,
                             ttl: int = 300) -> Optional[dict]:
        """Retrieve a cached response if it exists and has not expired."""
        row = self.conn.execute(
            "SELECT response_json, cached_at FROM query_cache WHERE cache_key = ?",
            (cache_key,)
        ).fetchone()
        if row and (time.time() - row["cached_at"]) < ttl:
            return json.loads(row["response_json"])
        return None

    def get_local_articles(self, category: Optional[str] = None,
                            limit: int = 25) -> list:
        """Query the local article cache."""
        sql = "SELECT * FROM articles"
        params = []
        if category:
            sql += " WHERE category = ?"
            params.append(category)
        sql += " ORDER BY published_date DESC LIMIT ?"
        params.append(limit)

        rows = self.conn.execute(sql, params).fetchall()
        return [dict(row) for row in rows]

    def close(self):
        """Close the database connection."""
        self.conn.close()

Now integrate the cache with the client. Here is a wrapper function that checks the cache before calling the API:

import hashlib
import json
from newsclient import NewsMeshClient
from cache import ArticleCache


def fetch_with_cache(client: NewsMeshClient, cache: ArticleCache,
                     method: str = "latest", ttl: int = 300, **kwargs) -> dict:
    """Fetch articles with local caching.

    Args:
        client: The NewsMesh API client.
        cache: The article cache instance.
        method: API method to call ("trending", "latest", or "search").
        ttl: Cache time-to-live in seconds (default: 5 minutes).
        **kwargs: Arguments passed to the API method.

    Returns:
        The API response dict, either from cache or fresh.
    """
    # Build a unique cache key from the method and parameters
    key_data = json.dumps({"method": method, **kwargs}, sort_keys=True)
    cache_key = hashlib.sha256(key_data.encode()).hexdigest()

    # Check cache first
    cached = cache.get_cached_response(cache_key, ttl=ttl)
    if cached:
        return cached

    # Call the API
    if method == "trending":
        response = client.get_trending(**kwargs)
    elif method == "search":
        response = client.search(**kwargs)
    else:
        response = client.get_latest(**kwargs)

    # Cache the response and individual articles
    cache.cache_query_response(cache_key, response, ttl=ttl)
    if "data" in response:
        cache.cache_articles(response["data"])

    return response

With this caching layer, repeated requests for the same query within the TTL window return instantly from SQLite instead of making a network call. A five-minute TTL is a reasonable default for development. For a news aggregator that updates every 15 minutes, you could increase the TTL to 900 seconds. The key insight is that news does not change second-by-second for most use cases, so even a short cache dramatically reduces load while keeping content fresh enough for your users.

The get_local_articles method also lets you query your cached articles directly, which is useful for building offline features or running local analytics across previously fetched data.

Step 4: Create a CLI Interface

A command-line interface is the fastest way to interact with your aggregator during development, and it remains useful as a standalone tool for power users who prefer the terminal. The CLI we will build supports all three API methods (trending, latest, search) with flags for every filter, a verbose mode that shows article descriptions and metadata, and an option to bypass the cache when you need fresh results.

Python's argparse module makes it straightforward to build a well-structured CLI with help text, argument validation, and subcommands.

Create cli.py:

import argparse
import sys
from newsclient import NewsMeshClient
from cache import ArticleCache


def display_articles(articles: list, verbose: bool = False):
    """Print articles to the terminal."""
    if not articles:
        print("No articles found.")
        return

    for i, article in enumerate(articles, 1):
        print(f"\n{'='*60}")
        print(f"  {i}. {article['title']}")
        print(f"     Source: {article['source']}  |  "
              f"Category: {article.get('category', 'N/A')}")
        print(f"     Date: {article.get('published_date', 'N/A')}")

        if verbose:
            desc = article.get("description", "")
            if desc:
                # Wrap long descriptions
                print(f"     {desc[:200]}{'...' if len(desc) > 200 else ''}")
            topics = article.get("topics", [])
            if topics:
                print(f"     Topics: {', '.join(topics)}")
            people = article.get("people", [])
            if people:
                print(f"     People: {', '.join(people)}")

        print(f"     Link: {article.get('link', 'N/A')}")

    print(f"\n{'='*60}")
    print(f"Showing {len(articles)} article(s).")


def main():
    parser = argparse.ArgumentParser(
        description="News Aggregator - Fetch and filter news from NewsMesh"
    )
    parser.add_argument(
        "command",
        choices=["trending", "latest", "search"],
        help="What to fetch: trending, latest, or search"
    )
    parser.add_argument(
        "--api-key",
        required=True,
        help="Your NewsMesh API key"
    )
    parser.add_argument(
        "--limit", type=int, default=10,
        help="Number of articles to fetch (default: 10)"
    )
    parser.add_argument(
        "--category",
        help="Filter by category (e.g., technology, politics, business)"
    )
    parser.add_argument(
        "--country",
        help="Filter by country code (e.g., us, gb, de)"
    )
    parser.add_argument(
        "--query", "-q",
        help="Search query (required for 'search' command)"
    )
    parser.add_argument(
        "--from-date",
        help="Start date filter (YYYY-MM-DD)"
    )
    parser.add_argument(
        "--to-date",
        help="End date filter (YYYY-MM-DD)"
    )
    parser.add_argument(
        "--verbose", "-v", action="store_true",
        help="Show descriptions, topics, and people"
    )
    parser.add_argument(
        "--no-cache", action="store_true",
        help="Bypass the local cache"
    )

    args = parser.parse_args()

    # Validate search command requires a query
    if args.command == "search" and not args.query:
        parser.error("The 'search' command requires --query / -q")

    client = NewsMeshClient(args.api_key)
    cache = ArticleCache()

    try:
        # Build keyword arguments for the API call
        kwargs = {"limit": args.limit}
        if args.category:
            kwargs["category"] = args.category
        if args.country:
            kwargs["country"] = args.country
        if args.from_date:
            kwargs["from_date"] = args.from_date
        if args.to_date:
            kwargs["to_date"] = args.to_date

        if args.command == "search":
            kwargs["query"] = args.query
            if args.no_cache:
                response = client.search(**kwargs)
            else:
                response = fetch_with_cache(client, cache,
                                            method="search", **kwargs)
        elif args.command == "trending":
            if args.no_cache:
                response = client.get_trending(limit=args.limit)
            else:
                response = fetch_with_cache(client, cache,
                                            method="trending",
                                            limit=args.limit)
        else:
            if args.no_cache:
                response = client.get_latest(**kwargs)
            else:
                response = fetch_with_cache(client, cache,
                                            method="latest", **kwargs)

        display_articles(response.get("data", []), verbose=args.verbose)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    finally:
        cache.close()


# Import fetch_with_cache from our caching module
from cache import ArticleCache
import hashlib
import json

def fetch_with_cache(client, cache, method="latest", ttl=300, **kwargs):
    key_data = json.dumps({"method": method, **kwargs}, sort_keys=True)
    cache_key = hashlib.sha256(key_data.encode()).hexdigest()

    cached = cache.get_cached_response(cache_key, ttl=ttl)
    if cached:
        return cached

    if method == "trending":
        response = client.get_trending(**kwargs)
    elif method == "search":
        response = client.search(**kwargs)
    else:
        response = client.get_latest(**kwargs)

    cache.cache_query_response(cache_key, response, ttl=ttl)
    if "data" in response:
        cache.cache_articles(response["data"])
    return response


if __name__ == "__main__":
    main()

Here are some example commands:

# Fetch trending articles
python cli.py trending --api-key YOUR_KEY --limit 5

# Latest technology news with details
python cli.py latest --api-key YOUR_KEY --category technology --verbose

# Search for a specific topic in a country
python cli.py search --api-key YOUR_KEY -q "electric vehicles" --country de --limit 10

# Articles from a specific date range, skip cache
python cli.py latest --api-key YOUR_KEY --from-date 2026-01-20 --to-date 2026-01-25 --no-cache

The CLI is already a fully functional news aggregator at this point. You can fetch trending stories to see what is making headlines globally, pull the latest articles filtered by category and country to build a focused reading list, or search for specific topics across the full article archive. Combined with the SQLite cache, it runs quickly even on repeated queries. The next step makes this accessible through a browser.

Step 5: Add a Web Frontend

A browser-based interface makes your aggregator accessible to non-technical users and is the natural starting point for any consumer-facing news product. The web frontend reuses the same client and cache modules you have already built, demonstrating the benefit of the modular architecture. We will use Flask for its simplicity and minimal boilerplate, but the same approach translates directly to FastAPI, Django, or any other Python web framework.

Create app.py:

from flask import Flask, render_template, request
from newsclient import NewsMeshClient
from cache import ArticleCache

app = Flask(__name__)

API_KEY = "your-api-key-here"
client = NewsMeshClient(API_KEY)
cache = ArticleCache()

CATEGORIES = [
    "politics", "technology", "business", "science",
    "health", "sports", "entertainment", "environment", "world"
]


@app.route("/")
def index():
    """Main page showing filtered news articles."""
    category = request.args.get("category", "")
    country = request.args.get("country", "")
    query = request.args.get("q", "")
    page_cursor = request.args.get("cursor", "")

    kwargs = {"limit": 20}
    method = "latest"

    if category:
        kwargs["category"] = category
    if country:
        kwargs["country"] = country
    if page_cursor:
        kwargs["cursor"] = page_cursor

    if query:
        method = "search"
        kwargs["query"] = query
        kwargs["sort_by"] = "relevant"

    # Fetch articles (with caching)
    import hashlib, json
    key_data = json.dumps({"method": method, **kwargs}, sort_keys=True)
    cache_key = hashlib.sha256(key_data.encode()).hexdigest()

    response = cache.get_cached_response(cache_key, ttl=300)
    if not response:
        if method == "search":
            response = client.search(**kwargs)
        else:
            response = client.get_latest(**kwargs)
        cache.cache_query_response(cache_key, response)
        if "data" in response:
            cache.cache_articles(response["data"])

    articles = response.get("data", [])
    next_cursor = response.get("next_cursor")

    return render_template(
        "index.html",
        articles=articles,
        categories=CATEGORIES,
        selected_category=category,
        selected_country=country,
        search_query=query,
        next_cursor=next_cursor,
    )


if __name__ == "__main__":
    app.run(debug=True, port=5001)

Create the templates directory and add templates/index.html:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>News Aggregator</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI",
               Roboto, sans-serif; background: #f5f5f5; color: #333;
               line-height: 1.6; }
        .container { max-width: 900px; margin: 0 auto; padding: 20px; }
        h1 { margin-bottom: 20px; }

        /* Filter bar */
        .filters { display: flex; gap: 10px; margin-bottom: 24px;
                   flex-wrap: wrap; }
        .filters select, .filters input, .filters button {
            padding: 8px 12px; border: 1px solid #ddd; border-radius: 6px;
            font-size: 14px; }
        .filters button { background: #111; color: #fff; cursor: pointer;
                          border: none; }

        /* Article cards */
        .article-card { background: #fff; border-radius: 8px; padding: 20px;
                        margin-bottom: 16px; box-shadow: 0 1px 3px
                        rgba(0,0,0,0.08); }
        .article-card h2 { font-size: 18px; margin-bottom: 6px; }
        .article-card h2 a { color: #111; text-decoration: none; }
        .article-card h2 a:hover { text-decoration: underline; }
        .article-meta { font-size: 13px; color: #777; margin-bottom: 8px; }
        .article-meta span { margin-right: 12px; }
        .article-desc { font-size: 15px; color: #555; }
        .topics { margin-top: 8px; }
        .topic-tag { display: inline-block; background: #eee;
                     border-radius: 4px; padding: 2px 8px; font-size: 12px;
                     margin-right: 4px; color: #555; }

        /* Pagination */
        .pagination { text-align: center; margin-top: 20px; }
        .pagination a { display: inline-block; padding: 10px 20px;
                        background: #111; color: #fff; border-radius: 6px;
                        text-decoration: none; }
    </style>
</head>
<body>
    <div class="container">
        <h1>News Aggregator</h1>

        <form class="filters" method="get" action="/">
            <select name="category">
                <option value="">All categories</option>
                {% for cat in categories %}
                <option value="{{ cat }}"
                    {{ 'selected' if cat == selected_category }}>
                    {{ cat|capitalize }}
                </option>
                {% endfor %}
            </select>

            <input type="text" name="country" placeholder="Country (e.g. us)"
                   value="{{ selected_country }}" style="width: 130px;">

            <input type="text" name="q" placeholder="Search keywords..."
                   value="{{ search_query }}" style="width: 220px;">

            <button type="submit">Filter</button>
        </form>

        {% for article in articles %}
        <div class="article-card">
            <h2><a href="{{ article.link }}" target="_blank" rel="noopener">
                {{ article.title }}
            </a></h2>
            <div class="article-meta">
                <span>{{ article.source }}</span>
                <span>{{ article.category }}</span>
                <span>{{ article.published_date[:10] }}</span>
            </div>
            {% if article.description %}
            <p class="article-desc">{{ article.description[:250] }}
                {{ '...' if article.description|length > 250 }}</p>
            {% endif %}
            {% if article.topics %}
            <div class="topics">
                {% for topic in article.topics[:5] %}
                <span class="topic-tag">{{ topic }}</span>
                {% endfor %}
            </div>
            {% endif %}
        </div>
        {% endfor %}

        {% if not articles %}
        <p>No articles found. Try adjusting your filters.</p>
        {% endif %}

        {% if next_cursor %}
        <div class="pagination">
            <a href="/?category={{ selected_category }}&country={{ selected_country }}&q={{ search_query }}&cursor={{ next_cursor }}">
                Load more articles
            </a>
        </div>
        {% endif %}
    </div>
</body>
</html>

Run the Flask app:

python app.py

Open http://localhost:5001 in your browser. You will see a clean list of article cards with a filter bar at the top. Each card displays the article headline as a clickable link to the original source, the publisher name, category, publication date, a truncated description, and topic tags. Select a category from the dropdown, enter a country code, or type a search query and click Filter to narrow the results. The "Load more articles" link at the bottom uses cursor-based pagination to fetch the next page without losing your current filter selections.

The HTML template is intentionally minimal. It uses only inline CSS and standard HTML form elements, making it easy to understand and customize. In a production application, you would want to replace this with a proper frontend framework or a CSS library like Tailwind, but for a tutorial and prototype this keeps the focus on the data flow.

Production Considerations

The Flask app above is suitable for local development. Before deploying to production, you should:

1. Move the API key to an environment variable. Replace the hardcoded key with os.environ.get("NEWSMESH_API_KEY") and set it in your server environment.
2. Use a production WSGI server. Run Flask behind Gunicorn or uWSGI instead of the development server: gunicorn app:app --bind 0.0.0.0:8000.
3. Add error handling. Wrap API calls in try/except blocks and show user-friendly error messages when the API is unreachable.
4. Upgrade to the official SDK. For production use, install the NewsMesh Python SDK with pip install newsmesh. It handles retries, pagination, error classes, and async support out of the box.

Next Steps

You now have a working news aggregator with API connectivity, multi-parameter filtering, local caching, a command-line interface, and a web frontend. That is a solid foundation, but there are many ways to extend it into a more powerful tool. Here are some directions to explore:

Add sentiment analysis. Analyze the tone of articles to surface positive, negative, or neutral coverage of a topic. Our guide on news sentiment analysis with Python walks through this using NLTK and transformer models.

Build an AI-powered digest. Use the aggregator as a data source for AI agents and RAG pipelines. Feed the cached articles into an LLM to generate daily summaries or answer questions about current events.

Set up automated fetching. Use Python's schedule library or a cron job to fetch articles on a timer and keep your local cache fresh without manual interaction:

import schedule
import time


def refresh_feeds():
    client = NewsMeshClient("your-api-key")
    cache = ArticleCache()
    for category in ["technology", "business", "politics"]:
        response = client.get_latest(limit=25, category=category)
        cache.cache_articles(response.get("data", []))
    cache.close()
    print("Feeds refreshed.")


schedule.every(15).minutes.do(refresh_feeds)

while True:
    schedule.run_pending()
    time.sleep(1)

Deploy to a server. Package the Flask app in a Docker container or deploy it to a platform like Railway, Render, or a VPS. Pair it with a PostgreSQL database instead of SQLite for production-grade caching. A typical Dockerfile would install your Python dependencies, copy the project files, and run Gunicorn as the entrypoint.

Add email or Slack notifications. Extend the scheduled fetching to detect new articles matching specific criteria and send alerts. For example, you could monitor mentions of your company name and push a Slack message whenever a new article appears.

Explore more use cases. See our use cases page for ideas on how teams in media, fintech, PR, and research are using news APIs to build production applications.

Conclusion

In this tutorial, you built a Python news aggregator from the ground up. Starting from an empty project directory, you created a reusable API client class that connects to the NewsMesh API and exposes clean methods for trending, latest, and search endpoints. You added multi-parameter filtering by category, country, date range, and keyword search, giving you fine-grained control over the articles you retrieve. You implemented a local SQLite cache with configurable TTL to reduce API usage and improve response times. You built an argparse-based command-line interface for quick terminal access, and finally you added a Flask web frontend with filter controls and cursor-based pagination for browser-based browsing.

The complete project is around 400 lines of Python across five files. Every component is modular and depends only on the layer below it, so you can swap in a different database backend, add new API endpoints, switch to async with aiohttp, or replace the frontend framework without rewriting the rest of the codebase.

The approach we followed here -- wrapping an API in a client class, adding a caching layer, and building interfaces on top -- is the same pattern used in production news products, monitoring tools, and research platforms. The difference between a tutorial project and a production system is mostly operational: error handling, logging, deployment, and scaling.

If you do not have a NewsMesh API key yet, sign up for free to get started. The development tier gives you access to all endpoints so you can build and test your aggregator without any upfront cost. When you are ready for real-time data and higher rate limits, paid plans start at $29/month. Full API documentation is available at newsmesh.co/docs.