Chapter 26: Building Your Own REST API

For 25 chapters, you've been the client. You've consumed APIs from OpenWeather, NewsAPI, Spotify, GitHub. You've sent HTTP requests, parsed JSON responses, handled authentication flows, and built applications on top of external data sources.

Basically, You've written code that calls requests.get() and processes what comes back.

Now you're ready for the flip side: you're going to be the API. You're going to build the system that receives those HTTP requests, validates authentication, enforces rate limits, queries databases, and returns JSON responses. You're going to create the endpoints that other developers integrate with. This isn't just a technical skill change. It's a professional identity shift.

The shift from API consumer to API producer: from making requests to receiving them

This chapter teaches you to build production-grade REST APIs using FastAPI, PostgreSQL, and professional patterns for authentication, rate limiting, and error handling.

You'll create a News Aggregator API that consolidates multiple news sources into one clean interface, demonstrating the exact patterns companies use to expose their data to developers. By the end, you'll have a deployed API with automatic documentation, secure authentication, and comprehensive test coverage—ready to show recruiters as proof you understand both sides of the API equation.

As a consumer, you think: "How do I call this API? What parameters does it accept? How do I handle errors?" These are client-side concerns. As a producer, the questions change completely.

What if someone makes 1,000 requests per second?

Without rate limiting, a single user can exhaust your database connections, spike your hosting costs, or crash your server. As a consumer, you never worried about this. As a producer, it's a critical design decision from day one.

How do you prevent abuse while staying accessible?

Every API needs authentication to track usage and prevent unauthorized access. But too much friction discourages adoption. Professional APIs balance security with developer experience—making authentication simple but secure.

How do you version your API without breaking existing users?

You'll improve your API over time, adding features and optimizing responses. But users depend on your current endpoints. Changing response formats breaks their code. Professional APIs version endpoints so improvements don't destroy existing integrations.

How do you help developers understand what went wrong?

As a consumer, you've seen generic 500 errors that provided no useful information. Frustrating. As a producer, you control what error messages users see. Good APIs return clear error messages, proper status codes, and actionable guidance. Poor APIs leave developers guessing.

Following news means checking multiple sources. Tech news from Hacker News, world events from The Guardian, breaking stories from NewsAPI. Each source has different API formats, different authentication requirements, different rate limits, and different response structures. Developers who want comprehensive news coverage must integrate with three, four, or five separate APIs.

Your News Aggregator API solves this problem. It aggregates articles from NewsAPI and The Guardian into one unified interface. Developers get one API key, one consistent response format, and access to multiple news sources through clean REST endpoints. Behind the scenes, your API handles the complexity: fetching from multiple sources, normalizing different response formats, caching results in PostgreSQL, and enforcing rate limits.

For consumers, the interface is simple:

GET /articles?category=technology&source=newsapi
Authorization: Bearer YOUR_API_KEY

Response:
{
  "articles": [
    {
      "id": "news_12345",
      "title": "AI Breakthrough in Medical Imaging",
      "description": "Researchers develop algorithm...",
      "url": "https://example.com/article",
      "source": "newsapi",
      "category": "technology",
      "published_at": "2024-12-09T10:30:00Z"
    }
  ],
  "total": 42,
  "page": 1
}

For you as the builder, the system demonstrates professional patterns:

Multi-source integration: Your API fetches from NewsAPI and Guardian, handles their different authentication methods, normalizes responses, and merges results into one clean format.

Database caching: Articles are stored in PostgreSQL with timestamps. Recent requests return cached data instantly. Stale cache triggers fresh API calls. This reduces external API usage and improves response times.

API key authentication: Users generate API keys through an admin endpoint. Keys are hashed before storage (like passwords). Every request validates the API key using middleware before processing.

Rate limiting: Each API key has usage limits tracked in PostgreSQL. Exceed your limit and you get a 429 status code with clear messaging about when limits reset.

Automatic documentation: FastAPI generates interactive API docs at /docs. Developers can test endpoints directly in the browser, see request/response schemas, and understand authentication requirements without reading separate documentation.

FastAPI is a Python framework for building high-performance APIs with built-in support for validation, documentation, and async.

Python has several frameworks for building APIs:

• Flask (simple but requires extensions for validation and docs)
• Django REST Framework (powerful but heavyweight)
• FastAPI (modern with built-in features for production APIs)

For this chapter, FastAPI is the clear choice.

Automatic OpenAPI documentation

Write your endpoints and FastAPI generates interactive documentation automatically at /docs. No separate documentation to maintain. No Swagger configuration files. The docs update automatically when you change code.

Built-in Pydantic validation

Define your request and response schemas using Python type hints. FastAPI automatically validates incoming data, generates clear error messages for invalid input, and provides type safety throughout your codebase. No manual validation code needed.

Async support for performance

FastAPI is built on modern async Python, allowing high-performance concurrent request handling. While this chapter uses synchronous code for clarity, you can add async support later when you need it.

Industry adoption

Microsoft, Uber, and Netflix use FastAPI in production. It's not an experimental framework—it's proven technology with active maintenance and strong community support.

Developer experience

FastAPI's automatic error messages, clear documentation, and Pydantic integration make debugging significantly easier than manual validation frameworks. When something goes wrong, you get specific, actionable error messages.

By the end of this chapter, you'll be able to:

• Design RESTful APIs following industry conventions for resource naming, HTTP methods, and status codes
• Build production-ready APIs with FastAPI, Pydantic validation, and automatic OpenAPI documentation
• Implement API key authentication with secure hashing and middleware-based validation
• Integrate PostgreSQL databases with APIs using SQLAlchemy for data persistence and caching
• Add rate limiting to prevent abuse and track API usage per key
• Test API endpoints comprehensively using pytest with authentication and database mocking
• Deploy authenticated APIs to production with environment-based configuration
• Create developer-friendly API documentation that makes your API easy to adopt

We'll build the News Aggregator API systematically, adding complexity progressively:

By Section 7, you'll have a complete, working API. Sections 8-9 polish it into a production-ready, portfolio-worthy project.

The next section lays the conceptual foundation. Before writing any FastAPI code, you'll learn REST principles, HTTP methods, status codes, and resource naming conventions. Understanding these fundamentals ensures your API follows industry standards from the start.

REST stands for Representational State Transfer, but that formal definition obscures the practical meaning. REST is a set of design conventions that make APIs intuitive and predictable. When your API follows REST principles, developers who've used other REST APIs immediately understand how to use yours.

Resources, not actions

REST APIs model your data as resources (nouns) accessed through URLs. You have /articles not /getArticles. Resources are things: articles, users, comments, categories. Actions (getting, creating, deleting) are conveyed through HTTP methods, not URL names.

HTTP methods convey intent

The HTTP method tells the server what you want to do with a resource. GET /articles retrieves articles. POST /articles creates a new article. DELETE /articles/123 removes article 123. The URL identifies the resource. The method specifies the action.

Stateless requests

Each request contains all information needed to process it. The server doesn't remember previous requests from this client. Authentication credentials are sent with every request. Query parameters specify filtering. The server has no "session" tracking what this client did before. This makes REST APIs scalable—any server can handle any request without shared state.

Predictable structure

REST APIs follow consistent patterns. /articles returns multiple articles. /articles/123 returns one specific article. /categories/technology/articles returns articles in the technology category. Once developers learn your patterns, they can predict other endpoints without reading documentation.

HTTP defines several request methods, but REST APIs primarily use five. Each method has specific semantics that convey intent without looking at request bodies or responses.

Safe methods

Safe methods (GET) guarantee no side effects. Calling GET /articles doesn't modify the database, charge credit cards, or send emails. Browsers and proxies can safely cache GET responses and retry failed GET requests without risk.

Idempotent methods

Idempotent methods (GET, PUT, PATCH, DELETE) produce the same result when called multiple times. Calling DELETE /articles/123 twice still deletes article 123 once—the second call finds nothing to delete and returns 204 anyway. This makes these methods safe to retry on network failures.

Non-idempotent methods

Non-idempotent methods (POST) create different results on repeated calls. Each POST /articles creates a new article, so calling it twice creates two articles. Clients should be cautious about retrying POST requests after timeouts.

HTTP status codes communicate outcomes without requiring clients to parse error messages. Professional APIs use status codes correctly and consistently. Here are the essential codes for REST APIs:

URL structure communicates your API's organization. Well-named resources make your API intuitive. Poorly named resources create confusion and require extensive documentation.

Use plural nouns for collections

/articles represents the collection of all articles. GET /articles retrieves multiple articles. POST /articles creates a new article in the collection. Consistent pluralization makes endpoints predictable.

Use IDs for specific resources

/articles/123 identifies article 123. GET /articles/123 retrieves this specific article. DELETE /articles/123 removes it. The ID in the URL clearly indicates you're operating on one specific resource, not the collection.

Nest related resources logically

/categories/technology/articles retrieves articles in the technology category. The URL structure mirrors the relationship: articles belong to categories. This makes the hierarchy clear without requiring query parameters.

Use query parameters for filtering

/articles?source=guardian&category=technology filters articles by source and category. Query parameters are perfect for optional filters, sorting, and pagination. The base resource remains /articles while query parameters refine what's returned.

Keep URLs lowercase with hyphens

/api-keys is standard REST convention. Not /ApiKeys or /api_keys. Hyphens separate words in URLs. Underscores can be hard to see when URLs are underlined in documentation.

GET    /articles                     # List all articles
GET    /articles/123                 # Get specific article
GET    /articles?category=tech       # Filter articles
GET    /categories                   # List categories
GET    /categories/tech/articles     # Articles in category
POST   /articles                     # Create new article
PUT    /articles/123                 # Replace article 123
DELETE /articles/123                 # Delete article 123

GET    /getArticles                  # Action in URL (should be GET /articles)
GET    /article_list                 # Underscore and non-standard naming
POST   /createArticle                # Action in URL (should be POST /articles)
GET    /articles/get/123             # Redundant 'get' in path
GET    /ARTICLES                     # Uppercase (violates conventions)
GET    /fetch-tech-news-items        # Overly specific, not RESTful

Before writing code, design your API's interface. List every endpoint, the HTTP method it uses, what parameters it accepts, and what it returns. This design-first approach ensures consistent patterns and prevents you from building endpoints that don't align with REST principles.

For the News Aggregator API, here's the complete interface design:

# Article Endpoints
GET    /articles                   # List articles (paginated)
  Query params: 
    - category: Filter by category (optional)
    - source: Filter by source (newsapi, guardian) (optional)
    - page: Page number for pagination (default: 1)
    - limit: Results per page (default: 20, max: 100)
  Returns: List of articles with pagination metadata

GET    /articles/{article_id}      # Get specific article
  Returns: Single article object or 404

GET    /articles/search            # Search articles by keyword
  Query params:
    - q: Search query (required)
    - category: Filter by category (optional)
  Returns: Matching articles

# Category and Source Endpoints
GET    /categories                 # List available categories
  Returns: Array of category names

GET    /sources                    # List available sources
  Returns: Array of source identifiers

# Admin Endpoints (require admin API key)
POST   /admin/api-keys             # Generate new API key
  Body: {"name": "key description", "tier": "basic"}
  Returns: {"api_key": "generated_key", "key_id": 123}

DELETE /admin/api-keys/{key_id}    # Revoke API key
  Returns: 204 No Content

# Health Check
GET    /health                     # Health check endpoint
  Returns: {"status": "healthy", "timestamp": "..."}

This interface follows REST conventions: plural nouns for collections, path parameters for IDs, query parameters for filters, appropriate HTTP methods, and logical grouping. With this design documented, implementation becomes straightforward.

Create a new directory for the News Aggregator API and set up a virtual environment. FastAPI requires Python 3.10+ for modern type hints.

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

pip install fastapi uvicorn[standard] sqlalchemy psycopg2-binary python-dotenv requests

fastapi: The web framework for building APIs. Provides routing, validation, and documentation generation.

uvicorn[standard]: ASGI server for running FastAPI applications. The [standard] includes optimized dependencies.

sqlalchemy: SQL toolkit and ORM for database operations. Handles PostgreSQL connections and query building.

psycopg2-binary: PostgreSQL adapter for Python. Required for SQLAlchemy to connect to PostgreSQL.

python-dotenv: Loads environment variables from .env files. Essential for managing secrets in development and production.

requests: HTTP library for calling external APIs (NewsAPI and Guardian). You're already familiar with this from earlier chapters.

Create main.py with a minimal FastAPI application to verify everything works.

from fastapi import FastAPI
from datetime import datetime

app = FastAPI(
    title="News Aggregator API",
    description="Unified interface for multiple news sources",
    version="1.0.0"
)

@app.get("/")
def root():
    return {
        "message": "News Aggregator API",
        "version": "1.0.0",
        "docs_url": "/docs"
    }

@app.get("/health")
def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat()
    }

Lines 4-8: Initialize FastAPI with metadata. This appears in the auto-generated documentation, helping users understand what your API does.

Lines 10-16: The @app.get("/") decorator registers a route. When someone requests GET /, FastAPI calls this function and returns its result as JSON automatically.

Lines 18-22: Health check endpoint. Standard pattern for monitoring API availability. Returns current timestamp to verify the server is responding.

Run the server with uvicorn:

uvicorn main:app --reload

# Output:
# INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
# INFO:     Started reloader process
# INFO:     Application startup complete.

main:app tells uvicorn to look for the app object in main.py.

--reload enables auto-reload. Uvicorn watches for file changes and restarts the server automatically. Perfect for development, but never use in production.

Your API is now running on http://localhost:8000. Visit that URL in your browser and you'll see the welcome message JSON. Visit http://localhost:8000/health for the health check response.

Automatic Documentation

Now visit http://localhost:8000/docs. FastAPI automatically generated interactive API documentation:

FastAPI's automatic documentation at /docs. Click any endpoint to test it directly in the browser.

This documentation updates automatically when you add endpoints or change response types. You never write or maintain separate API documentation—FastAPI generates it from your code.

Path parameters extract values from the URL path. They're perfect for resource IDs: /articles/123 has article ID 123 as a path parameter.

from fastapi import FastAPI, HTTPException

app = FastAPI()

# Simulated database
articles_db = {
    1: {"id": 1, "title": "FastAPI Tutorial", "source": "internal"},
    2: {"id": 2, "title": "REST API Design", "source": "internal"},
    3: {"id": 3, "title": "PostgreSQL Basics", "source": "internal"}
}

@app.get("/articles/{article_id}")
def get_article(article_id: int):
    if article_id not in articles_db:
        raise HTTPException(status_code=404, detail="Article not found")
    return articles_db[article_id]

Line 13: The {article_id} curly braces in the path define a path parameter. FastAPI extracts whatever appears in that position from the URL.

Line 14: The function parameter article_id: int receives the extracted value. The : int type hint tells FastAPI to convert the string from the URL into an integer. If the URL contains /articles/abc, FastAPI automatically returns a 422 validation error because "abc" isn't a valid integer.

Lines 15-16: HTTPException is FastAPI's way to return error responses. When you raise HTTPException(status_code=404, detail="Article not found"), FastAPI converts it into a proper HTTP 404 response with the detail message in JSON.

Test the endpoint:

curl http://localhost:8000/articles/1
# Response: {"id":1,"title":"FastAPI Tutorial","source":"internal"}

curl http://localhost:8000/articles/999
# Response: {"detail":"Article not found"}

curl http://localhost:8000/articles/abc
# Response: {"detail":[{"type":"int_parsing","loc":["path","article_id"],"msg":"Input should be a valid integer..."}]}

Notice FastAPI's validation: requesting /articles/abc returns a detailed error message explaining that article_id must be an integer. You didn't write validation code—FastAPI generated it from the type hint.

Query parameters provide optional filtering and configuration. They appear after the ? in URLs: /articles?category=tech&limit=10.

from typing import Optional

@app.get("/articles")
def list_articles(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20
):
    # Start with all articles
    results = list(articles_db.values())
    
    # Apply filters if provided
    if category:
        results = [a for a in results if a.get("category") == category]
    if source:
        results = [a for a in results if a.get("source") == source]
    
    # Apply limit
    results = results[:limit]
    
    return {
        "articles": results,
        "count": len(results),
        "filters": {"category": category, "source": source, "limit": limit}
    }

Line 5: Optional[str] = None makes category optional. If the URL doesn't include ?category=..., the parameter defaults to None.

Line 7: limit: int = 20 provides a default value. If the URL omits ?limit=..., it defaults to 20.

Lines 13-16: Filter the results based on which query parameters were provided. This demonstrates the pattern: accept optional parameters, apply filters conditionally.

Test with various query combinations:

# All articles
curl http://localhost:8000/articles

# Filter by source
curl http://localhost:8000/articles?source=internal

# Combine filters
curl http://localhost:8000/articles?source=internal&limit=2

FastAPI automatically validates query parameter types. If you request ?limit=abc, you get a validation error explaining that limit must be an integer.

Pydantic models define the shape of request bodies and responses. They provide automatic validation, type conversion, and documentation generation. This is FastAPI's superpower: type-safe APIs without manual validation code.

from pydantic import BaseModel, Field
from datetime import datetime

class ArticleCreate(BaseModel):
    """Schema for creating new articles."""
    title: str = Field(..., min_length=1, max_length=500)
    description: str
    url: str
    source: str
    category: str

class ArticleResponse(BaseModel):
    """Schema for article responses."""
    id: int
    title: str
    description: str
    url: str
    source: str
    category: str
    published_at: datetime
    created_at: datetime

@app.post("/articles", response_model=ArticleResponse, status_code=201)
def create_article(article: ArticleCreate):
    # Generate ID (in real app, database does this)
    new_id = max(articles_db.keys()) + 1 if articles_db else 1
    
    # Create article with timestamps
    new_article = {
        "id": new_id,
        **article.model_dump(),
        "published_at": datetime.utcnow(),
        "created_at": datetime.utcnow()
    }
    
    articles_db[new_id] = new_article
    return new_article

Lines 4-10: ArticleCreate defines what data clients send when creating articles. Field(...) marks title as required with length constraints.

Lines 12-21: ArticleResponse defines what the API returns. It includes additional fields like id and timestamps that clients don't provide.

Line 23: response_model=ArticleResponse tells FastAPI to validate the function's return value against this schema and include it in docs. status_code=201 returns "Created" status for successful POST requests.

Line 24: article: ArticleCreate tells FastAPI to parse the request body as JSON and validate it against ArticleCreate. If validation fails, FastAPI returns a 422 error with specific details about what's wrong.

Test creating an article:

# Valid request
curl -X POST http://localhost:8000/articles \
  -H "Content-Type: application/json" \
  -d '{
    "title": "New FastAPI Article",
    "description": "Learn FastAPI quickly",
    "url": "https://example.com/fastapi",
    "source": "internal",
    "category": "technology"
  }'
# Response: {"id":4,"title":"New FastAPI Article",...}

# Invalid request (missing required field)
curl -X POST http://localhost:8000/articles \
  -H "Content-Type: application/json" \
  -d '{"title": "Incomplete Article"}'
# Response: {"detail":[{"type":"missing","loc":["body","description"],"msg":"Field required"}]}

The invalid request returns a clear error: description field is required but missing. FastAPI generated this validation automatically from the Pydantic model.

You've used SQLite in previous chapters for the Music Time Machine. SQLite works perfectly for single-user applications and development. But production APIs need PostgreSQL for several critical reasons:

Concurrent writes

SQLite locks the entire database file during writes. Only one writer at a time. If your API has 10 concurrent requests trying to write, they queue sequentially. PostgreSQL handles thousands of concurrent writes through its sophisticated locking system. Each table row can be locked independently.

Connection pooling

PostgreSQL runs as a separate server process. Multiple API servers connect to one PostgreSQL instance. SQLite is a file that each process opens directly. You can't share an SQLite database between multiple API servers without complex file locking that doesn't work reliably.

Data integrity

PostgreSQL enforces foreign key constraints, check constraints, and complex transaction isolation levels. SQLite supports these but with limitations. For production data that multiple services depend on, PostgreSQL's robustness matters.

Scalability

PostgreSQL can grow to terabytes with replication, read replicas, and horizontal scaling. SQLite is designed for embedded use—one file on one machine. When your API outgrows one server, PostgreSQL scales. SQLite doesn't.

For local development, install PostgreSQL and create a database for the News Aggregator API. In production (Section 9), you'll use Railway's managed PostgreSQL.

# macOS (Homebrew)
brew install postgresql@15
brew services start postgresql@15

# Ubuntu/Debian
sudo apt-get update
sudo apt-get install postgresql postgresql-contrib

# Create database and user
createdb news_aggregator
psql news_aggregator -c "CREATE USER api_user WITH PASSWORD 'dev_password';"
psql news_aggregator -c "GRANT ALL PRIVILEGES ON DATABASE news_aggregator TO api_user;"

For Windows: Download PostgreSQL installer from postgresql.org, run it, and use pgAdmin to create the database and user.

Create a .env file to store your database connection string:

DATABASE_URL=postgresql://api_user:dev_password@localhost:5432/news_aggregator
SECRET_KEY=your-secret-key-for-hashing
NEWSAPI_KEY=your-newsapi-key
GUARDIAN_KEY=your-guardian-key

Important: Add .env to .gitignore. Never commit secrets to version control.

SQLAlchemy is Python's most popular ORM (Object-Relational Mapping) library. It lets you define database tables as Python classes, and converts Python operations into SQL queries automatically.

Create database.py for all database-related code:

import os
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

# Database connection string from environment
DATABASE_URL = os.getenv("DATABASE_URL")

# Create engine with connection pooling
engine = create_engine(
    DATABASE_URL,
    pool_size=5,          # 5 persistent connections
    max_overflow=10,      # Up to 10 additional connections under load
    pool_pre_ping=True    # Verify connections before using
)

# Session factory for database operations
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for all models
Base = declarative_base()


# Database Models
class Article(Base):
    """Cached news articles."""
    __tablename__ = "articles"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(500), nullable=False)
    description = Column(String(2000))
    url = Column(String(500), unique=True, nullable=False)
    source = Column(String(100), nullable=False)
    category = Column(String(100), nullable=False)
    published_at = Column(DateTime, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)


class APIKey(Base):
    """API keys for authentication."""
    __tablename__ = "api_keys"
    
    id = Column(Integer, primary_key=True, index=True)
    key_hash = Column(String(64), unique=True, nullable=False, index=True)
    name = Column(String(200))
    rate_limit_tier = Column(String(50), default="basic")
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    last_used_at = Column(DateTime, nullable=True)
    
    # Relationship to usage logs
    usage_logs = relationship("UsageLog", back_populates="api_key")


class UsageLog(Base):
    """Track API usage for rate limiting."""
    __tablename__ = "usage_logs"
    
    id = Column(Integer, primary_key=True, index=True)
    api_key_id = Column(Integer, ForeignKey("api_keys.id"), nullable=False, index=True)
    endpoint = Column(String(200), nullable=False)
    method = Column(String(10), nullable=False)
    status_code = Column(Integer, nullable=False)
    timestamp = Column(DateTime, default=datetime.utcnow, index=True)
    
    # Relationship to API key
    api_key = relationship("APIKey", back_populates="usage_logs")


# Dependency injection for database sessions
def get_db():
    """Get database session for request handlers."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# Create all tables
def init_db():
    """Initialize database tables."""
    Base.metadata.create_all(bind=engine)

Lines 14-19: Create database engine with connection pooling. pool_size=5 maintains 5 persistent connections. max_overflow=10 allows up to 15 total connections under load. pool_pre_ping=True tests connections before use—if PostgreSQL restarted, SQLAlchemy detects dead connections and creates new ones.

Lines 28-41: Article model represents cached news articles. url is unique to prevent duplicate articles. updated_at with onupdate automatically updates the timestamp on any modification.

Lines 44-56: APIKey model stores hashed API keys. key_hash is indexed for fast lookups. rate_limit_tier determines request limits per hour. is_active allows key revocation without deletion.

Lines 59-69: UsageLog tracks every API request for rate limiting and analytics. timestamp is indexed for efficient time-based queries.

Lines 74-80: get_db() is a dependency injection function. FastAPI calls it, gets a database session, passes it to your endpoint handler, and closes it automatically when the handler returns. This ensures connections are properly closed even if exceptions occur.

Initialize the database tables:

from database import init_db

if __name__ == "__main__":
    print("Creating database tables...")
    init_db()
    print("Tables created successfully!")

python init_db.py
# Output: Creating database tables...
# Output: Tables created successfully!

Verify the tables were created:

psql news_aggregator -c "\dt"
#           List of relations
#  Schema |    Name     | Type  |   Owner   
# --------+-------------+-------+-----------
#  public | api_keys    | table | api_user
#  public | articles    | table | api_user
#  public | usage_logs  | table | api_user

Now update main.py to use the PostgreSQL database instead of the in-memory dictionary.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import Optional
from pydantic import BaseModel
from datetime import datetime

from database import get_db, Article as DBArticle

app = FastAPI(
    title="News Aggregator API",
    description="Unified interface for multiple news sources",
    version="1.0.0"
)


# Pydantic models for API
class ArticleResponse(BaseModel):
    """Schema for article responses."""
    id: int
    title: str
    description: str
    url: str
    source: str
    category: str
    published_at: datetime
    created_at: datetime
    
    class Config:
        from_attributes = True  # Allow conversion from SQLAlchemy models


@app.get("/articles", response_model=list[ArticleResponse])
def list_articles(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20,
    db: Session = Depends(get_db)
):
    # Build query
    query = db.query(DBArticle)
    
    # Apply filters
    if category:
        query = query.filter(DBArticle.category == category)
    if source:
        query = query.filter(DBArticle.source == source)
    
    # Apply limit and execute
    articles = query.limit(limit).all()
    
    return articles


@app.get("/articles/{article_id}", response_model=ArticleResponse)
def get_article(article_id: int, db: Session = Depends(get_db)):
    article = db.query(DBArticle).filter(DBArticle.id == article_id).first()
    
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    
    return article

Line 7: Import Article as DBArticle to distinguish database models from Pydantic models.

Lines 16-27: ArticleResponse Pydantic model for API responses. class Config: from_attributes = True lets Pydantic convert SQLAlchemy models directly to Pydantic models.

Line 35: db: Session = Depends(get_db) injects a database session. FastAPI calls get_db(), gets a session, passes it to this function, and closes it after the function returns.

Lines 37-46: Build a SQL query using SQLAlchemy's query builder. db.query(DBArticle) starts the query. .filter() adds WHERE clauses. .limit() adds LIMIT. .all() executes and returns results.

Line 53: .first() returns one result or None. More efficient than .all()[0] when you need exactly one row.

Test the database-backed endpoints. They work identically to the in-memory version, but data persists across server restarts.

Without connection pooling, each request would open a new database connection, execute queries, and close the connection. Opening connections is expensive—requires TCP handshake, authentication, and PostgreSQL process initialization. High-traffic APIs would waste time connecting instead of serving requests.

Connection pooling maintains a pool of open connections. When a request needs database access, it borrows a connection from the pool. After the request completes, the connection returns to the pool for reuse. This eliminates connection overhead and limits total connections to prevent overwhelming the database.

The configuration in database.py sets pool_size=5 (5 persistent connections) and max_overflow=10 (up to 10 additional connections during traffic spikes). This means 5 connections stay open always, and the pool can grow to 15 connections under load. After traffic subsides, overflow connections close and the pool shrinks back to 5.

pool_pre_ping=True tests connections before using them. If PostgreSQL restarted or a connection went stale, SQLAlchemy detects this and creates a fresh connection automatically. Without pre-ping, your API would attempt to use dead connections and return errors.

Public APIs without authentication face inevitable abuse. Automated scrapers hammer endpoints with thousands of requests per minute. Malicious users exploit open APIs to consume resources, test vulnerabilities, or extract data for resale. Without authentication, you can't track usage, enforce rate limits, or prevent abuse.

Authentication serves three purposes: identification (who is making this request?), access control (are they allowed to access this endpoint?), and usage tracking (how many requests have they made?). API keys accomplish all three with minimal implementation complexity.

Remember Chapter 2 when you got your first API key from NewsAPI? You registered, received a key, and included it in every request header. That simple flow—generate key, distribute to users, validate on every request—is what this section implements. The system that issued your NewsAPI key is exactly what you're building now.

API keys are credentials like passwords. Store them hashed, never in plain text. When users generate keys, you show the key once, they copy it, and you store only the hash. If your database leaks, attackers get hashes, not working keys.

import secrets
import hashlib
from datetime import datetime
from sqlalchemy.orm import Session
from database import APIKey as DBAPIKey


def generate_api_key() -> str:
    """Generate a random API key."""
    return secrets.token_urlsafe(32)  # 32 bytes = 256 bits of randomness


def hash_api_key(api_key: str) -> str:
    """Hash an API key using SHA-256."""
    return hashlib.sha256(api_key.encode()).hexdigest()


def create_api_key(db: Session, name: str, tier: str = "basic") -> dict:
    """
    Create a new API key.
    Returns dict with key and id. Store the hash, return the key once.
    """
    # Generate random key
    raw_key = generate_api_key()
    key_hash = hash_api_key(raw_key)
    
    # Store hash in database
    db_key = DBAPIKey(
        key_hash=key_hash,
        name=name,
        rate_limit_tier=tier,
        is_active=True
    )
    db.add(db_key)
    db.commit()
    db.refresh(db_key)
    
    # Return the raw key ONCE - user must save it
    return {
        "api_key": raw_key,  # Show this once, never again
        "key_id": db_key.id,
        "name": name,
        "tier": tier,
        "created_at": db_key.created_at
    }


def validate_api_key(db: Session, api_key: str) -> DBAPIKey | None:
    """
    Validate an API key and return the associated key object.
    Returns None if invalid or inactive.
    """
    key_hash = hash_api_key(api_key)
    
    # Find key by hash
    db_key = db.query(DBAPIKey).filter(
        DBAPIKey.key_hash == key_hash,
        DBAPIKey.is_active == True
    ).first()
    
    if db_key:
        # Update last used timestamp
        db_key.last_used_at = datetime.utcnow()
        db.commit()
    
    return db_key

Line 9: secrets.token_urlsafe(32) generates cryptographically secure random strings. 32 bytes = 256 bits of entropy. This produces keys like xQz4K8m_NpD1... that are URL-safe (no special characters needing encoding).

Line 14: SHA-256 hashing is one-way. Given a hash, computing the original key is computationally infeasible. Attackers who steal your database get hashes they can't use.

Lines 24-25: Generate key, hash it immediately. The raw key exists briefly in memory, gets returned to the user once, then disappears forever.

Lines 38-43: Return the raw key to the user. This is their only chance to copy it. After this response, only the hash exists in your database. You cannot recover the original key.

Lines 51-59: Validation hashes the provided key and looks up the hash in the database. If found and active, validation succeeds. This is how every request gets authenticated.

Middleware intercepts requests before they reach endpoint handlers. Authentication middleware checks every request for a valid API key. Invalid or missing keys get rejected with 401 before your endpoint code runs.

from fastapi import FastAPI, HTTPException, Depends, Header
from sqlalchemy.orm import Session
from typing import Optional

from database import get_db, APIKey as DBAPIKey
from auth import validate_api_key

app = FastAPI()


def require_api_key(
    authorization: Optional[str] = Header(None),
    db: Session = Depends(get_db)
) -> DBAPIKey:
    """
    Dependency that validates API key from Authorization header.
    Raises 401 if key is missing or invalid.
    """
    if not authorization:
        raise HTTPException(
            status_code=401,
            detail="Missing Authorization header",
            headers={"WWW-Authenticate": "Bearer"}
        )
    
    # Extract Bearer token
    if not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=401,
            detail="Invalid Authorization format. Use: Bearer YOUR_API_KEY"
        )
    
    api_key = authorization.replace("Bearer ", "")
    
    # Validate key
    db_key = validate_api_key(db, api_key)
    if not db_key:
        raise HTTPException(
            status_code=401,
            detail="Invalid or inactive API key"
        )
    
    return db_key


# Public endpoints (no authentication)
@app.get("/")
def root():
    return {"message": "News Aggregator API", "docs": "/docs"}


# Protected endpoints (require authentication)
@app.get("/articles", response_model=list[ArticleResponse])
def list_articles(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20,
    api_key: DBAPIKey = Depends(require_api_key),
    db: Session = Depends(get_db)
):
    # Authentication succeeded - api_key contains validated key object
    query = db.query(DBArticle)
    
    if category:
        query = query.filter(DBArticle.category == category)
    if source:
        query = query.filter(DBArticle.source == source)
    
    articles = query.limit(limit).all()
    return articles

Lines 11-23: require_api_key is a dependency function. authorization: Optional[str] = Header(None) extracts the Authorization header. If missing, raise 401.

Lines 26-31: Validate header format. API keys should be sent as Authorization: Bearer YOUR_API_KEY. This is the standard format used by GitHub, Stripe, and most modern APIs.

Lines 35-41: Call validate_api_key() to check if the key exists and is active. If validation fails, raise 401 with clear error message.

Line 58: Add api_key: DBAPIKey = Depends(require_api_key) to any endpoint that requires authentication. FastAPI runs require_api_key before the endpoint handler. If authentication fails, the endpoint never runs.

Test authentication:

# Request without authentication
curl http://localhost:8000/articles
# Response: {"detail":"Missing Authorization header"}

# Request with invalid key
curl -H "Authorization: Bearer invalid_key" http://localhost:8000/articles
# Response: {"detail":"Invalid or inactive API key"}

# First, generate a valid key (requires admin endpoint)
curl -X POST http://localhost:8000/admin/api-keys \
  -H "Content-Type: application/json" \
  -d '{"name": "Test Key", "tier": "basic"}'
# Response: {"api_key":"xQz4K8m_NpD1...","key_id":1}

# Request with valid key
curl -H "Authorization: Bearer xQz4K8m_NpD1..." http://localhost:8000/articles
# Response: {"articles": [...], "count": 10}

Notice how authentication failures return clear error messages with appropriate status codes. Professional APIs communicate exactly what went wrong.

Create admin endpoints for generating and revoking API keys. In production, protect these with additional authentication (admin credentials). For development, they're open for convenience.

from pydantic import BaseModel
from auth import create_api_key


class APIKeyCreate(BaseModel):
    """Request body for creating API keys."""
    name: str
    tier: str = "basic"


class APIKeyResponse(BaseModel):
    """Response with API key (shown once)."""
    api_key: str
    key_id: int
    name: str
    tier: str
    created_at: datetime


@app.post("/admin/api-keys", response_model=APIKeyResponse, status_code=201)
def generate_api_key(
    key_data: APIKeyCreate,
    db: Session = Depends(get_db)
):
    """Generate a new API key. Returns key once - user must save it."""
    result = create_api_key(
        db=db,
        name=key_data.name,
        tier=key_data.tier
    )
    return result


@app.delete("/admin/api-keys/{key_id}", status_code=204)
def revoke_api_key(
    key_id: int,
    db: Session = Depends(get_db)
):
    """Revoke an API key (mark as inactive)."""
    db_key = db.query(DBAPIKey).filter(DBAPIKey.id == key_id).first()
    
    if not db_key:
        raise HTTPException(status_code=404, detail="API key not found")
    
    db_key.is_active = False
    db.commit()
    
    return None  # 204 responses have no body

Lines 19-31: Generate API key endpoint. Returns the key once. Emphasize to users: "Save this key now. You won't see it again."

Lines 34-47: Revoke endpoint sets is_active = False instead of deleting. This preserves usage history for analytics while preventing future API calls. Deleted keys lose all tracking data.

Without rate limiting, a single user can overwhelm your API. One aggressive script making 1,000 requests per second consumes server resources, exhausts database connections, and impacts performance for all users. Rate limiting protects your infrastructure and ensures fair resource distribution.

Rate limits also prevent accidental abuse. A developer's buggy script stuck in an infinite loop shouldn't crash your API. A misconfigured client retrying failed requests shouldn't exhaust your quota with external APIs. Rate limiting stops runaway processes before they cause damage.

Professional APIs publish rate limits clearly: "100 requests per hour for free tier, 1000 per hour for paid tier." Users understand the boundaries and build applications accordingly. When they exceed limits, they get clear 429 responses indicating when limits reset. This creates predictable, manageable API usage.

Fixed-window rate limiting divides time into fixed windows (e.g., one hour) and counts requests per window. If a user makes 100 requests in the current hour, request 101 gets blocked until the hour resets. This is simple to implement and reason about.

More sophisticated algorithms exist (sliding window, token bucket), but fixed-window provides good protection with minimal complexity. It's what most APIs use, including the ones you've worked with throughout this book.

from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from database import UsageLog, APIKey as DBAPIKey

# Rate limit tiers (requests per hour)
RATE_LIMITS = {
    "basic": 100,
    "premium": 1000,
    "unlimited": float('inf')
}


def check_rate_limit(db: Session, api_key: DBAPIKey) -> tuple[bool, dict]:
    """
    Check if API key has exceeded rate limit.
    Returns (allowed: bool, info: dict with details)
    """
    tier = api_key.rate_limit_tier
    limit = RATE_LIMITS.get(tier, 100)
    
    # Calculate current window (start of current hour)
    now = datetime.utcnow()
    window_start = now.replace(minute=0, second=0, microsecond=0)
    
    # Count requests in current window
    request_count = db.query(UsageLog).filter(
        UsageLog.api_key_id == api_key.id,
        UsageLog.timestamp >= window_start
    ).count()
    
    # Calculate when window resets
    window_reset = window_start + timedelta(hours=1)
    seconds_until_reset = int((window_reset - now).total_seconds())
    
    allowed = request_count < limit
    
    info = {
        "limit": limit,
        "remaining": max(0, limit - request_count),
        "reset": window_reset.isoformat(),
        "reset_in_seconds": seconds_until_reset
    }
    
    return allowed, info


def log_request(db: Session, api_key_id: int, endpoint: str, method: str, status_code: int):
    """Log API request for rate limiting and analytics."""
    log_entry = UsageLog(
        api_key_id=api_key_id,
        endpoint=endpoint,
        method=method,
        status_code=status_code
    )
    db.add(log_entry)
    db.commit()

Lines 6-10: Define rate limit tiers. Basic users get 100 requests per hour, premium get 1000, unlimited tiers have no limit. Adjust these based on your infrastructure capacity and business model.

Lines 22-23: Calculate window start by truncating to the beginning of the current hour. If it's 15:37:22, window_start becomes 15:00:00.

Lines 26-29: Query usage_logs table for requests by this API key since window start. Count tells us how many requests they've made in the current hour.

Lines 36-42: Build info dict with limit, remaining requests, and when the window resets. This gets returned in response headers so users know their quota status.

Integrate rate limiting into the authentication dependency so every protected endpoint checks limits automatically.

from fastapi import FastAPI, HTTPException, Depends, Header, Request, Response
from rate_limit import check_rate_limit, log_request


def require_api_key_with_rate_limit(
    request: Request,
    response: Response,
    authorization: Optional[str] = Header(None),
    db: Session = Depends(get_db)
) -> DBAPIKey:
    """
    Validate API key and check rate limit.
    Returns validated key object.
    Raises 401 for auth errors, 429 for rate limit exceeded.
    """
    # Validate authentication (same as before)
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid Authorization")
    
    api_key = authorization.replace("Bearer ", "")
    db_key = validate_api_key(db, api_key)
    if not db_key:
        raise HTTPException(status_code=401, detail="Invalid API key")
    
    # Check rate limit
    allowed, rate_info = check_rate_limit(db, db_key)
    
    # Add rate limit headers to response
    response.headers["X-RateLimit-Limit"] = str(rate_info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(rate_info["remaining"])
    response.headers["X-RateLimit-Reset"] = rate_info["reset"]
    
    if not allowed:
        raise HTTPException(
            status_code=429,
            detail=f"Rate limit exceeded. Resets in {rate_info['reset_in_seconds']} seconds",
            headers={
                "Retry-After": str(rate_info['reset_in_seconds']),
                "X-RateLimit-Limit": str(rate_info["limit"]),
                "X-RateLimit-Reset": rate_info["reset"]
            }
        )
    
    # Log request
    log_request(
        db=db,
        api_key_id=db_key.id,
        endpoint=request.url.path,
        method=request.method,
        status_code=200  # Endpoint execution will update if needed
    )
    
    return db_key


# Update endpoint to use new dependency
@app.get("/articles", response_model=list[ArticleResponse])
def list_articles(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20,
    api_key: DBAPIKey = Depends(require_api_key_with_rate_limit),
    db: Session = Depends(get_db)
):
    query = db.query(DBArticle)
    
    if category:
        query = query.filter(DBArticle.category == category)
    if source:
        query = query.filter(DBArticle.source == source)
    
    articles = query.limit(limit).all()
    return articles

Lines 6-10: Updated dependency now requires Request and Response objects. Request provides endpoint path and method. Response lets us add custom headers.

Lines 25-31: Check rate limit and add headers to every response. Clients see their quota status even on successful requests.

Lines 33-42: If rate limit exceeded, raise 429 with clear error message and Retry-After header telling clients when they can retry.

Lines 45-51: Log every request to usage_logs table. This data powers rate limiting and provides analytics on API usage.

Test rate limiting:

# Check headers on normal request
curl -i -H "Authorization: Bearer YOUR_KEY" http://localhost:8000/articles
# HTTP/1.1 200 OK
# X-RateLimit-Limit: 100
# X-RateLimit-Remaining: 99
# X-RateLimit-Reset: 2024-12-09T16:00:00

# Script to hit rate limit (basic tier = 100 requests/hour)
for i in {1..101}; do
  curl -H "Authorization: Bearer YOUR_KEY" http://localhost:8000/articles
done

# Request 101 returns:
# HTTP/1.1 429 Too Many Requests
# Retry-After: 1847
# X-RateLimit-Limit: 100
# X-RateLimit-Reset: 2024-12-09T16:00:00
# {"detail":"Rate limit exceeded. Resets in 1847 seconds"}

The 429 response includes both human-readable error message and machine-readable headers. Client applications can parse Retry-After to implement exponential backoff automatically.

The News Aggregator API fetches articles from NewsAPI and The Guardian, normalizes their different response formats, caches results in PostgreSQL, and returns a unified response. This section brings together authentication, rate limiting, database operations, and external API integration.

The integration challenge

NewsAPI and The Guardian have completely different response structures. NewsAPI returns articles in an articles array with fields like publishedAt. The Guardian uses response.results with webPublicationDate. Your API must normalize both formats into one consistent structure.

Caching strategy

External API calls are slow and consume your quota. Cache articles in PostgreSQL with timestamps. Recent articles (less than 1 hour old) return from cache instantly. Stale articles trigger fresh API calls. This reduces external API usage by 80-90% while keeping content reasonably fresh.

Error handling

External APIs fail. NewsAPI might be down, The Guardian might rate limit you, network timeouts occur. Your API must handle these gracefully. If NewsAPI fails, try The Guardian. If both fail, return cached articles with a warning header. Never crash because an external dependency is unavailable.

Create sources.py to handle external API calls and response normalization.

import os
import requests
from datetime import datetime
from typing import List, Dict, Optional


def fetch_newsapi(category: Optional[str] = None, limit: int = 20) -> List[Dict]:
    """
    Fetch articles from NewsAPI.
    Returns normalized article dictionaries.
    """
    api_key = os.getenv("NEWSAPI_KEY")
    if not api_key:
        raise ValueError("NEWSAPI_KEY not configured")
    
    url = "https://newsapi.org/v2/top-headlines"
    params = {
        "apiKey": api_key,
        "country": "us",
        "pageSize": min(limit, 100)
    }
    
    if category:
        params["category"] = category
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # Normalize NewsAPI response format
        articles = []
        for article in data.get("articles", []):
            normalized = {
                "title": article.get("title", ""),
                "description": article.get("description", ""),
                "url": article.get("url", ""),
                "source": "newsapi",
                "category": category or "general",
                "published_at": parse_timestamp(article.get("publishedAt")),
            }
            articles.append(normalized)
        
        return articles
    
    except requests.RequestException as e:
        print(f"NewsAPI error: {e}")
        return []  # Graceful degradation


def fetch_guardian(category: Optional[str] = None, limit: int = 20) -> List[Dict]:
    """
    Fetch articles from The Guardian.
    Returns normalized article dictionaries.
    """
    api_key = os.getenv("GUARDIAN_KEY")
    if not api_key:
        raise ValueError("GUARDIAN_KEY not configured")
    
    url = "https://content.guardianapis.com/search"
    params = {
        "api-key": api_key,
        "page-size": min(limit, 50),
        "show-fields": "headline,trailText,shortUrl"
    }
    
    if category:
        params["section"] = category
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # Normalize Guardian response format
        articles = []
        results = data.get("response", {}).get("results", [])
        for article in results:
            fields = article.get("fields", {})
            normalized = {
                "title": fields.get("headline", article.get("webTitle", "")),
                "description": fields.get("trailText", ""),
                "url": fields.get("shortUrl", article.get("webUrl", "")),
                "source": "guardian",
                "category": article.get("sectionName", category or "general"),
                "published_at": parse_timestamp(article.get("webPublicationDate")),
            }
            articles.append(normalized)
        
        return articles
    
    except requests.RequestException as e:
        print(f"Guardian API error: {e}")
        return []


def parse_timestamp(timestamp_str: Optional[str]) -> datetime:
    """
    Parse ISO 8601 timestamp from various formats.
    Returns datetime object or current time if parsing fails.
    """
    if not timestamp_str:
        return datetime.utcnow()
    
    try:
        # Handle ISO 8601 with timezone
        if "T" in timestamp_str:
            clean_ts = timestamp_str.replace("Z", "+00:00")
            return datetime.fromisoformat(clean_ts.split("+")[0])
        return datetime.utcnow()
    except (ValueError, AttributeError):
        return datetime.utcnow()


def fetch_from_all_sources(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20
) -> List[Dict]:
    """
    Fetch from multiple sources and merge results.
    If source specified, fetch only from that source.
    """
    articles = []
    
    if source == "newsapi" or source is None:
        newsapi_articles = fetch_newsapi(category, limit)
        articles.extend(newsapi_articles)
    
    if source == "guardian" or source is None:
        guardian_articles = fetch_guardian(category, limit)
        articles.extend(guardian_articles)
    
    # Sort by published date (newest first)
    articles.sort(key=lambda x: x["published_at"], reverse=True)
    
    # Apply limit after merging
    return articles[:limit]

Lines 7-44: fetch_newsapi() calls NewsAPI's top headlines endpoint. The response structure has articles in data["articles"] with fields like title, publishedAt. We normalize to our standard format.

Lines 31-39: Normalization converts NewsAPI's format to your standard structure. Every article gets title, description, url, source, category, published_at regardless of original format.

Lines 42-44: Graceful degradation: if NewsAPI fails, return empty list rather than crashing. Users still get Guardian results. Log the error for monitoring.

Lines 47-90: fetch_guardian() follows the same pattern but handles Guardian's different response structure. Articles are in response.results with metadata in nested fields objects.

Lines 93-110: parse_timestamp() handles different timestamp formats from different APIs. ISO 8601 with timezones, without timezones, or missing timestamps all get parsed consistently.

Lines 113-135: fetch_from_all_sources() orchestrates multi-source fetching. If user specifies source, fetch only from that source. Otherwise fetch from all sources, merge results, sort by date, and apply limit.

Every call to external APIs costs time and counts against rate limits. Caching stores recent articles in PostgreSQL. If cached data is fresh (less than 1 hour old), serve it immediately. If stale, fetch fresh data and update cache.

from datetime import datetime, timedelta
from sqlalchemy.orm import Session
from database import Article as DBArticle
from sources import fetch_from_all_sources
from typing import List, Optional


# Cache freshness: 1 hour
CACHE_TTL_MINUTES = 60


def get_or_fetch_articles(
    db: Session,
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = 20
) -> List[DBArticle]:
    """
    Main caching logic: return cached data if fresh, otherwise fetch and cache.
    This is the function your endpoints call.
    """
    # Check cache freshness
    cutoff_time = datetime.utcnow() - timedelta(minutes=CACHE_TTL_MINUTES)
    
    # Build query for cached articles
    query = db.query(DBArticle).filter(DBArticle.created_at >= cutoff_time)
    
    if category:
        query = query.filter(DBArticle.category == category)
    if source:
        query = query.filter(DBArticle.source == source)
    
    # Get cached articles
    cached = query.order_by(DBArticle.published_at.desc()).limit(limit).all()
    
    # If we have enough fresh articles, return from cache
    if len(cached) >= limit:
        print(f"Cache hit: {category}/{source}")
        return cached
    
    # Cache miss or insufficient articles - fetch fresh data
    print(f"Cache miss: {category}/{source} - fetching from sources")
    fresh_articles = fetch_from_all_sources(category, source, limit)
    
    if not fresh_articles:
        # External APIs failed - return stale cache if it exists
        print("External APIs failed - returning stale cache")
        stale_query = db.query(DBArticle)
        if category:
            stale_query = stale_query.filter(DBArticle.category == category)
        if source:
            stale_query = stale_query.filter(DBArticle.source == source)
        return stale_query.order_by(DBArticle.published_at.desc()).limit(limit).all()
    
    # Cache fresh articles
    cached_articles = []
    for article_data in fresh_articles:
        # Check if article already exists (by URL)
        existing = db.query(DBArticle).filter(
            DBArticle.url == article_data["url"]
        ).first()
        
        if existing:
            # Update existing article's timestamp to keep it fresh
            existing.updated_at = datetime.utcnow()
            existing.title = article_data["title"]
            existing.description = article_data["description"]
            cached_articles.append(existing)
        else:
            # Insert new article
            new_article = DBArticle(
                title=article_data["title"],
                description=article_data["description"],
                url=article_data["url"],
                source=article_data["source"],
                category=article_data["category"],
                published_at=article_data["published_at"]
            )
            db.add(new_article)
            cached_articles.append(new_article)
    
    db.commit()
    
    # Refresh objects to get generated IDs
    for article in cached_articles:
        db.refresh(article)
    
    return cached_articles

Lines 8-9: Cache TTL (Time To Live) is 1 hour. Articles older than 1 hour are considered stale. Adjust this based on your needs: breaking news might use 5-15 minutes, evergreen content might use several hours.

Lines 12-39: Main caching logic. Check database for articles created within the last hour. If sufficient fresh articles exist, return immediately without external API calls. This is the 80-90% cache hit rate in action.

Lines 42-52: Cache miss path. Fetch from external APIs. If they fail, fall back to stale cache (better to show slightly old news than nothing). Graceful degradation prevents total failures.

Lines 55-80: Cache population with upsert logic. If article URL already exists, update its timestamp to keep it fresh. If new, insert. This prevents duplicate articles while keeping cache current.

Now update main.py with the complete articles endpoint that integrates everything: authentication, rate limiting, caching, and multi-source fetching.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from typing import Optional, List
from pydantic import BaseModel, Field
from datetime import datetime

from database import get_db, Article as DBArticle
from auth import require_api_key_with_rate_limit, APIKey as DBAPIKey
from cache import get_or_fetch_articles

app = FastAPI(
    title="News Aggregator API",
    description="Unified interface for multiple news sources",
    version="1.0.0"
)


# Pydantic response models
class ArticleResponse(BaseModel):
    """Individual article in API responses."""
    id: int
    title: str
    description: str
    url: str
    source: str
    category: str
    published_at: datetime
    
    class Config:
        from_attributes = True


class ArticleListResponse(BaseModel):
    """Paginated article list response."""
    articles: List[ArticleResponse]
    total: int
    cache_status: str


@app.get("/articles", response_model=ArticleListResponse)
def list_articles(
    category: Optional[str] = None,
    source: Optional[str] = None,
    limit: int = Field(20, ge=1, le=100),
    api_key: DBAPIKey = Depends(require_api_key_with_rate_limit),
    db: Session = Depends(get_db)
):
    """
    List articles from multiple news sources with smart caching.
    
    Query Parameters:
    - category: Filter by category (technology, business, etc.)
    - source: Filter by source (newsapi, guardian, or omit for all)
    - limit: Results per page (1-100, default 20)
    
    Requires: Valid API key in Authorization: Bearer header
    Rate Limits: Basic tier = 100 requests/hour
    """
    # Get articles (from cache or fresh fetch)
    articles = get_or_fetch_articles(db, category, source, limit)
    
    # Determine cache status
    if articles:
        newest_cached = max(article.created_at for article in articles)
        age_minutes = (datetime.utcnow() - newest_cached).total_seconds() / 60
        cache_status = "hit" if age_minutes < 60 else "miss"
    else:
        cache_status = "empty"
    
    return {
        "articles": articles,
        "total": len(articles),
        "cache_status": cache_status
    }


@app.get("/articles/{article_id}", response_model=ArticleResponse)
def get_article(
    article_id: int,
    api_key: DBAPIKey = Depends(require_api_key_with_rate_limit),
    db: Session = Depends(get_db)
):
    """Get a specific article by ID."""
    article = db.query(DBArticle).filter(DBArticle.id == article_id).first()
    
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")
    
    return article

Lines 40-73: Main /articles endpoint. Uses get_or_fetch_articles() which handles all caching logic transparently. Returns metadata about cache status so clients can understand response times.

Lines 62-67: Calculate cache status by checking age of newest article. If less than 60 minutes old, it's a cache hit. This transparency helps consumers understand whether data is instant (cached) or took time (fresh fetch).

Lines 76-87: Single article retrieval by ID. Simple database lookup with 404 error if not found. Protected by same authentication and rate limiting as list endpoint.

Test the complete system:

# First request - cache miss, fetches from external APIs
curl -H "Authorization: Bearer YOUR_KEY" \
  "http://localhost:8000/articles?category=technology&limit=10"

# Response shows cache_status: "miss"
# Server logs: "Cache miss: technology/None - fetching from sources"

# Second request within 1 hour - cache hit
curl -H "Authorization: Bearer YOUR_KEY" \
  "http://localhost:8000/articles?category=technology&limit=10"

# Response shows cache_status: "hit"
# Server logs: "Cache hit: technology/None"
# Response is instant - no external API calls

# Filter by specific source
curl -H "Authorization: Bearer YOUR_KEY" \
  "http://localhost:8000/articles?source=guardian&limit=5"

# Get single article
curl -H "Authorization: Bearer YOUR_KEY" \
  "http://localhost:8000/articles/1"

Notice cache behavior: first request is slow (200-500ms) because it calls external APIs. Second identical request is instant (5-20ms) because it returns cached data. This is production caching in action.

APIs are contracts with other developers. Breaking changes destroy integrations without warning. Authentication bugs expose security vulnerabilities. Rate limiting failures allow abuse. Untested APIs are untrustworthy APIs.

Testing prevents regressions. You add a feature, accidentally break authentication, deploy to production, and every API call fails with 401 errors. Without tests, you discover this when users report problems. With tests, you catch it before committing code.

Professional APIs have comprehensive test suites covering authentication, authorization, rate limiting, input validation, error handling, and external API failures. This section builds that test suite for the News Aggregator API.

FastAPI provides excellent testing support through TestClient. Combined with pytest fixtures, you can test your entire API without running a real server.

Install testing dependencies:

pip install pytest pytest-cov httpx

Create conftest.py with pytest fixtures for database and authentication:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from main import app
from database import Base, get_db
from auth import create_api_key

# Test database (in-memory SQLite)
TEST_DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@pytest.fixture
def db():
    """Create test database and return session."""
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)


@pytest.fixture
def client(db):
    """Create test client with test database."""
    def override_get_db():
        try:
            yield db
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()


@pytest.fixture
def api_key(db):
    """Create test API key and return the raw key."""
    result = create_api_key(db, name="Test Key", tier="basic")
    return result["api_key"]


@pytest.fixture
def auth_headers(api_key):
    """Create authorization headers with API key."""
    return {"Authorization": f"Bearer {api_key}"}

Lines 10-13: In-memory SQLite for tests. Each test gets a fresh database that's destroyed after the test completes. Fast, isolated, no cleanup needed.

Lines 16-25: Database fixture. Creates tables, yields session to tests, drops tables on teardown. Every test starts with clean database state.

Lines 28-38: Test client fixture. Overrides get_db() dependency to use test database. Tests call endpoints through client.get(), client.post(), etc.

Lines 41-49: Authentication fixtures. api_key generates a valid key. auth_headers creates properly formatted authorization headers. Tests can use these without manual setup.

Create test_auth.py to verify authentication works correctly.

def test_authentication_required(client):
    """Test that protected endpoints require authentication."""
    response = client.get("/articles")
    assert response.status_code == 401
    assert "Missing Authorization" in response.json()["detail"]


def test_invalid_api_key(client):
    """Test that invalid API keys are rejected."""
    headers = {"Authorization": "Bearer invalid_key_12345"}
    response = client.get("/articles", headers=headers)
    assert response.status_code == 401
    assert "Invalid or inactive" in response.json()["detail"]


def test_valid_authentication(client, auth_headers):
    """Test that valid API keys are accepted."""
    response = client.get("/articles", headers=auth_headers)
    assert response.status_code == 200


def test_api_key_generation(client, db):
    """Test API key generation endpoint."""
    response = client.post(
        "/admin/api-keys",
        json={"name": "New Test Key", "tier": "premium"}
    )
    assert response.status_code == 201
    data = response.json()
    assert "api_key" in data
    assert data["tier"] == "premium"
    assert len(data["api_key"]) > 20

Lines 1-5: Test protected endpoints reject unauthenticated requests. Verifies 401 status and meaningful error message.

Lines 8-13: Test invalid keys are rejected. Ensures authentication doesn't accept arbitrary strings.

Lines 16-19: Test valid authentication succeeds. Uses auth_headers fixture for clean test code.

Lines 22-31: Test key generation returns proper format. Verifies response includes the raw key (shown once) and correct metadata.

Tests shouldn't call real external APIs. External APIs are slow, cost money, rate limit you, and can fail independently of your code. Mock them for fast, reliable tests.

from unittest.mock import patch, Mock


@patch('sources.requests.get')
def test_cache_behavior(mock_get, client, auth_headers, db):
    """Test that caching works correctly."""
    # Mock external API response
    mock_response = Mock()
    mock_response.json.return_value = {
        "articles": [
            {
                "title": "Test Article",
                "description": "Test description",
                "url": "https://example.com/test",
                "publishedAt": "2024-12-09T10:00:00Z"
            }
        ]
    }
    mock_response.raise_for_status.return_value = None
    mock_get.return_value = mock_response
    
    # First request (cache miss)
    response1 = client.get("/articles?category=technology", headers=auth_headers)
    assert response1.status_code == 200
    assert mock_get.called
    
    # Second request (cache hit - shouldn't call external API)
    mock_get.reset_mock()
    response2 = client.get("/articles?category=technology", headers=auth_headers)
    assert response2.status_code == 200
    assert not mock_get.called  # Cache hit - no external API call

Lines 4-20: Mock NewsAPI with @patch decorator. mock_get replaces requests.get(), returning controlled test data. Tests run instantly without network calls.

Lines 22-30: Test caching logic. First request calls external API. Second request returns cached data without calling API. Proves caching works.

Run tests with coverage:

# Run all tests
pytest

# Run with coverage report
pytest --cov=. --cov-report=html

# Output:
# ========================== test session starts ===========================
# collected 10 items
#
# test_auth.py::test_authentication_required PASSED              [ 10%]
# test_auth.py::test_invalid_api_key PASSED                      [ 20%]
# test_auth.py::test_valid_authentication PASSED                 [ 30%]
# test_auth.py::test_api_key_generation PASSED                   [ 40%]
# test_articles.py::test_cache_behavior PASSED                   [ 50%]
#
# ========================= 10 passed in 1.84s =============================
#
# Coverage report saved to htmlcov/index.html

Your API works locally. Now deploy it to Railway with PostgreSQL, proper environment configuration, and a live URL for your portfolio.

Create requirements.txt

Railway needs to know which packages to install:

fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
psycopg2-binary==2.9.9
python-dotenv==1.0.0
requests==2.31.0

Create Procfile

Railway uses Procfile to know how to start your application:

web: uvicorn main:app --host 0.0.0.0 --port $PORT

--host 0.0.0.0 binds to all network interfaces (required for Railway). $PORT uses Railway's assigned port.

Add startup event for database initialization

Update main.py to create tables on startup:

from database import init_db

@app.on_event("startup")
def startup_event():
    """Initialize database tables on startup."""
    init_db()
    print("Database tables initialized")

Railway provides free hosting with PostgreSQL, automatic HTTPS, and GitHub integration for continuous deployment.

Step 1: Create Railway Account and Project

1. Visit railway.app and sign up with GitHub
2. Click "New Project" → "Deploy from GitHub repo"
3. Authorize Railway to access your repositories
4. Select your News Aggregator API repository
5. Railway detects Python automatically and starts deployment

Step 2: Add PostgreSQL Database

1. In your Railway project, click "New" → "Database" → "Add PostgreSQL"
2. Railway creates a PostgreSQL instance and generates DATABASE_URL
3. The database URL is automatically added to your environment variables

Step 3: Configure Environment Variables

1. In your project, click "Variables"
2. Add NEWSAPI_KEY with your NewsAPI key
3. Add GUARDIAN_KEY with your Guardian key
4. Add SECRET_KEY with a random string
5. DATABASE_URL is already set by PostgreSQL service

Step 4: Verify Deployment

Railway provides a deployment URL like https://news-aggregator-production.up.railway.app. Test it:

# Test health check
curl https://your-app.up.railway.app/

# Generate API key
curl -X POST https://your-app.up.railway.app/admin/api-keys \
  -H "Content-Type: application/json" \
  -d '{"name": "Production Test", "tier": "basic"}'

# Test authenticated endpoint
curl -H "Authorization: Bearer YOUR_KEY" \
  https://your-app.up.railway.app/articles?category=technology

# View interactive docs
# Visit: https://your-app.up.railway.app/docs

Your API is now live with automatic HTTPS, PostgreSQL database, and interactive documentation. Every push to your GitHub main branch triggers automatic redeployment.

You've built a production REST API from scratch. You designed RESTful endpoints following industry conventions, implemented secure authentication with hashed API keys, added rate limiting to prevent abuse, integrated PostgreSQL for persistent storage, and deployed to Railway with automatic documentation. The News Aggregator API demonstrates professional backend development skills.

Think about the evolution: you started this book making simple GET requests to OpenWeather. Now you're building the servers that respond to those requests. You've gone from response = requests.get(url) to @app.get("/endpoint")—from client to server, from consumer to producer. This isn't just a technical skill change. It's a fundamental shift in how you understand web applications.

More importantly, you've completed the full circle. In Chapter 2, you made your first API request to NewsAPI. You received a JSON response and didn't think about the server behind it. Now you've built that server. You know how API keys are generated, how rate limits are enforced, why responses are structured certain ways, and what happens when external dependencies fail. You understand both sides of the HTTP transaction.

The patterns you learned—REST principles, authentication middleware, rate limiting strategies, caching logic, graceful degradation—transfer to any API you build. Whether you're building internal tools for your company, creating SaaS products, or developing microservices, these are the foundational patterns of professional API development.

But your News Aggregator API runs on Railway's free tier. It serves development traffic perfectly but isn't configured for production scale. Chapter 27: Docker and Containerization teaches you to package your API into Docker containers for consistent deployment across any environment. Chapter 28: AWS Production Deployment shows how to deploy containerized APIs to AWS with ECS Fargate, load balancing, and database replication. Chapter 29: Production Operations adds CI/CD pipelines, CloudWatch monitoring, auto-scaling, and blue-green deployments. Together, these chapters take your API from portfolio project to enterprise-scale infrastructure.

The progression is deliberate. Chapter 26 taught you to build APIs correctly. Chapters 27-29 teach you to scale them. By the end, you'll have the complete skill set: building production-ready APIs and deploying them at scale with professional DevOps practices. You're ready.