Python API Health Aggregator

You're staring at a terminal, manually curling five different services to see if they're up. Or worse, you've got five separate monitoring dashboards and you're tab-hopping between them. When something breaks, you waste precious minutes figuring out which service is down. This lesson solves that pain by showing you how to build a Python API health aggregator — a single script that checks multiple endpoints, reports their status in one place, and gives you a foundation for richer monitoring.

The problem this lesson solves

Operational teams often juggle dozens of microservices. Checking each one by hand is slow, error-prone, and doesn't scale. A health aggregator centralizes status checks into one command—or one scheduled job—so you can answer "Is everything up?" in seconds.

You could use a commercial tool like Datadog or Pingdom, but sometimes you need a lightweight, scriptable solution that's easy to extend. That's where Python shines: you can build a concise, dependency-light aggregator with just the standard library, drop it into your CI pipeline, or even run it as a cron job. This is a core DevOps skill: turning repetitive manual checks into repeatable automation.

Core concept / mental model

Think of your aggregator as a health dashboard in a script. You define a list of endpoints—like a phone book—and the script calls each one, measuring response time and HTTP status. It then categorizes each as healthy, degraded, or down, based on your rules.

Key components:

• Endpoint list: A collection of URLs (and optionally, expected status codes).
• Probe function: A function that performs the HTTP request and returns status info.
• Aggregation logic: Combines individual results into a summary and an overall verdict.
• Output formatter: Presents the results in a human- or machine-readable format.

Analogy: Imagine you're a security guard checking multiple doors. You walk to each door, try the handle, and note whether it's locked or open. The aggregator is your clipboard and summary report—it keeps track of every door and tells you the building is secure only if all doors are locked.

How it works step by step

The flow is straightforward:

1. Define your endpoints — typically in a list of dictionaries or a config file.
2. Probe each endpoint — use urllib.request (or requests if you prefer) to send a request with a timeout.
3. Record the result — capture HTTP status code, response time, and any errors.
4. Classify health — healthy if status is in the expected set (e.g., 200-299), degraded if response is slow, down if it errors or times out.
5. Aggregate and report — compute overall status (all healthy, some degraded, any down) and print a neat summary.

Let's see this in action.

Hands-on walkthrough

We'll build a script that monitors three example APIs. For realistic code, we'll use the standard library's urllib.request — no pip installs needed.

First, define your endpoints and a probe function:

import json
import time
import urllib.request
import urllib.error
from typing import Dict, List, Tuple

ENDPOINTS = [
    {"name": "GitHub API", "url": "https://api.github.com", "timeout": 5},
    {"name": "Example.com", "url": "https://example.com", "timeout": 5},
    {"name": "Nonexistent Service", "url": "https://this-domain-does-not-exist-12345.com", "timeout": 3},
]

def probe(url: str, timeout: int) -> Tuple[int, float, str]:
    """Return (status_code, response_time_seconds, error_message)."""
    start = time.monotonic()
    try:
        with urllib.request.urlopen(url, timeout=timeout) as resp:
            status = resp.getcode()
            return status, time.monotonic() - start, None
    except urllib.error.HTTPError as e:
        return e.code, time.monotonic() - start, None
    except Exception as e:
        return 0, time.monotonic() - start, str(e)

Next, add the classification and aggregation logic:

def classify(status: int, response_time: float, error: str) -> str:
    if error:
        return "down"
    if status >= 200 and status < 300:
        if response_time > 2.0:
            return "degraded"
        return "healthy"
    return "degraded"  # e.g., 3xx, 4xx, 5xx

def aggregate(results: List[Dict]) -> str:
    statuses = [r["health"] for r in results]
    if "down" in statuses:
        return "DOWN"
    if "degraded" in statuses:
        return "DEGRADED"
    return "UP"

def main() -> None:
    results = []
    for ep in ENDPOINTS:
        status_code, response_time, error = probe(ep["url"], ep["timeout"])
        health = classify(status_code, response_time, error)
        results.append({
            "name": ep["name"],
            "status_code": status_code,
            "response_time_ms": round(response_time * 1000),
            "health": health,
            "error": error,
        })
        print(f"{ep['name']}: {health} (HTTP {status_code}, {results[-1]['response_time_ms']}ms)")

    overall = aggregate(results)
    print(f"\nOverall status: {overall}")

if __name__ == "__main__":
    main()

When you run this script, you'll see output like:

GitHub API: healthy (HTTP 200, 234ms)
Example.com: healthy (HTTP 200, 87ms)
Nonexistent Service: down (HTTP 0, 0ms)

Overall status: DOWN

Pro tip: Use time.monotonic() instead of time.time() for measuring elapsed time—it's immune to system clock changes.

Now let's make this more flexible. We'll add support for expected status codes and a JSON output option for easy integration with other tools:

def probe_with_expected(url: str, expected_status: int, timeout: int) -> bool:
    """Return True if the endpoint returns the expected status and is reachable."""
    status, _, error = probe(url, timeout)
    return (error is None) and (status == expected_status)

# Example: check that GitHub returns 200 always
is_github_up = probe_with_expected("https://api.github.com", 200, 5)
print(f"GitHub API is up: {is_github_up}")

Do you need to run this on a schedule? Wrap the aggregation in a function and call it from a cron job or CI pipeline. For example, you could add a --json flag to print results as JSON:

import sys

def main() -> None:
    # ... results building code as above ...
    if "--json" in sys.argv:
        print(json.dumps(results, indent=2))
    else:
        for r in results:
            print(f"{r['name']}: {r['health']} (HTTP {r['status_code']}, {r['response_time_ms']}ms)")
        print(f"Overall: {aggregate(results)}")

This lets you pipe the output into tools like jq or other scripts.

Compare options / when to choose what

There are several ways to build an API health checker. Here's a quick comparison:

For a simple aggregator that runs on a schedule, the standard library is fine—it keeps your script portable. If you need to check hundreds of endpoints concurrently, httpx with asyncio is a better fit. For a one-off demonstration, requests gives you nicer code.

Pro tip: If you plan to extend this script into a full monitoring system, start with requests for readability—but abstract the HTTP call so you can swap libraries later without changing the aggregation logic.

Troubleshooting & edge cases

• Timeout errors: If an endpoint hangs, your script may wait indefinitely. Always set a timeout value (as shown). If a host is unreachable, you'll get a URLError; our generic exception handler catches it and marks the service as down. To distinguish DNS failures from connection refused, catch urllib.error.URLError and inspect e.reason.
• SSL certificate errors: Corporate or self-signed certificates can cause ssl.SSLError. For testing, you can add context=ssl._create_unverified_context() to urlopen()—but never do that in production. Better: add the custom CA to your environment or use requests with verify=.
• Redirects: By default, urlopen follows redirects (e.g., HTTP 301). This can mask issues. If you want to detect redirects as degraded, disable auto-follow (complex in urllib, easier in requests with allow_redirects=False).
• Local-only endpoints: If you're checking services on localhost, make sure they accept connections from your process. Check firewall rules and bind addresses.
• Concurrent checks: Running probes sequentially is slow for many endpoints. To speed up, use concurrent.futures.ThreadPoolExecutor—but watch out for rate limits and thread-safety.

What you learned & what's next

You now know how to build a Python API health aggregator: define endpoint lists, probe them with urllib, classify health, and produce a readable or JSON summary. You've applied the core concepts of automated status checking—a foundational DevOps pattern. You've also learned to handle timeouts, error codes, and offer different output formats.

Key takeaways: - Health checks should measure status and response time, not just connectivity. - Aggregation logic should produce an overall verdict that's easy to consume. - Standard library is enough for simple use; add dependencies only when needed. - Timeouts and error handling are critical to avoid stuck scripts. - Output formatting matters—JSON makes your aggregator composable.

Next step: Now that you can check APIs, you'll likely want to turn these checks into alerts. In the next lesson, you'll build a Python script that sends Slack notifications when an endpoint goes down—combining this aggregator with webhook automation. That's the natural evolution from detecting problems to reacting to them.