This tutorial guides you through building a command-line HTTP API health monitor dashboard using Python. It leverages requests for robust HTTP calls and ri
This tutorial guides you through building a command-line HTTP API health monitor dashboard using Python. It leverages requests for robust HTTP calls and rich for a beautiful, real-time console display.
In modern distributed systems, applications often rely on numerous external and internal HTTP APIs. Ensuring the continuous availability and responsiveness of these APIs is critical for the overall health of a service. Manual checks are tedious and impractical, while traditional monitoring systems can be overkill for simple, quick-glance checks during development or incident response.
This script addresses the need for a lightweight, real-time command-line tool that provides an immediate overview of multiple API endpoints' health. It helps developers and operations teams quickly identify which APIs are experiencing issues, such as downtime, high latency, or unexpected response codes, without needing to access complex dashboards or log files. This proactive monitoring can significantly reduce mean time to resolution (MTTR) for API-related incidents.
The HTTP API Health Monitor Dashboard operates on a simple, configuration-driven architecture. At its core, it reads a list of API endpoints and their expected behaviors from a YAML configuration file. For each defined API, the script performs a standard HTTP request (GET, POST, etc.) using the requests library, measures the response time, and compares the actual HTTP status code against the expected one.
The data flow begins with parsing the command-line arguments to locate the YAML configuration file. Once loaded, the script enters an infinite loop, periodically iterating through each API definition. For every API, it executes the HTTP check, captures the outcome (success/failure, latency, error message), and then aggregates these results. Finally, it uses the rich library to render a visually appealing, auto-updating table directly in the terminal, presenting the real-time status of all monitored APIs. Key design decisions include using requests for its comprehensive HTTP client features, PyYAML for flexible configuration, and rich for an enhanced user experience with live console updates and styled output.
#!/usr/bin/env python3
"""
HTTP API Health Monitor Dashboard
Monitors the health of multiple HTTP API endpoints defined in a YAML
configuration file, displaying their real-time status, latency, and
any encountered errors in a live, interactive console dashboard.
"""
import argparse
import os
import time
import requests
import yaml
from typing import Dict, Any, List, Optional
from rich.console import Console
from rich.table import Table
from rich.text import Text
from rich.live import Live
from rich.spinner import Spinner
from rich.panel import Panel
# --- Configuration Section ---
# Environment variables provide flexible configuration.
# Defaults are provided for convenience.
# Default path for the monitor configuration YAML file.
# Can be overridden by the --config argument or MONITOR_CONFIG_PATH env var.
DEFAULT_CONFIG_PATH: str = os.environ.get("MONITOR_CONFIG_PATH", "monitors.yaml")
# Default timeout for HTTP requests in seconds.
# Can be overridden by REQUEST_TIMEOUT env var or per-monitor config.
DEFAULT_REQUEST_TIMEOUT: float = float(os.environ.get("REQUEST_TIMEOUT", "5.0"))
# Default refresh interval for the dashboard in seconds.
# Can be overridden by REFRESH_INTERVAL env var.
DEFAULT_REFRESH_INTERVAL: float = float(os.environ.get("REFRESH_INTERVAL", "3.0"))
# --- Helper Functions ---
def load_monitors(config_path: str) -> List[Dict[str, Any]]:
"""
Loads API monitor configurations from a YAML file.
Args:
config_path: The path to the YAML configuration file.
Returns:
A list of dictionaries, each representing an API monitor.
Raises:
FileNotFoundError: If the config file does not exist.
yaml.YAMLError: If there's an error parsing the YAML file.
"""
try:
with open(config_path, 'r', encoding='utf-8') as f:
monitors_config = yaml.safe_load(f)
if not isinstance(monitors_config, list):
raise ValueError("Monitor configuration must be a list of API definitions.")
return monitors_config
except FileNotFoundError:
raise FileNotFoundError(f"Configuration file not found: '{config_path}'")
except yaml.YAMLError as e:
raise yaml.YAMLError(f"Error parsing YAML configuration: {e}")
except ValueError as e:
raise ValueError(f"Invalid configuration structure: {e}")
def check_api(monitor: Dict[str, Any]) -> Dict[str, Any]:
"""
Performs an HTTP request to the specified API endpoint and returns its status.
Args:
monitor: A dictionary containing API details (name, url, method, expected_status, etc.).
Returns:
A dictionary containing the check result (status, latency, message, color).
"""
name = monitor.get("name", "Unnamed API")
url = monitor.get("url")
method = monitor.get("method", "GET").upper()
expected_status = monitor.get("expected_status", 200)
headers = monitor.get("headers", {})
body = monitor.get("body")
timeout = monitor.get("timeout", DEFAULT_REQUEST_TIMEOUT)
if not url:
return {
"name": name,
"status": "❓ UNKNOWN",
"latency": "N/A",
"message": "Missing URL in configuration",
"color": "grey"
}
start_time = time.monotonic()
status_text = "❓ UNKNOWN"
message = "An unexpected error occurred."
color = "grey"
try:
response: requests.Response = requests.request(
method,
url,
headers=headers,
data=body,
timeout=timeout,
allow_redirects=True # Generally good practice for health checks
)
end_time = time.monotonic()
latency = round((end_time - start_time) * 1000, 2) # Latency in ms
if response.status_code == expected_status:
status_text = "✅ UP"
message = f"Status {response.status_code}"
color = "green"
else:
status_text = "❌ DOWN"
message = f"Expected {expected_status}, got {response.status_code}"
color = "red"
except requests.exceptions.Timeout:
latency = round((time.monotonic() - start_time) * 1000, 2)
status_text = "⚠️ TIMEOUT"
message = f"Request timed out after {timeout}s"
color = "yellow"
except requests.exceptions.ConnectionError as e:
latency = "N/A"
status_text = "❌ DOWN"
message = f"Connection error: {e}"
color = "red"
except requests.exceptions.HTTPError as e:
latency = "N/A"
status_text = "❌ DOWN"
message = f"HTTP error: {e}"
color = "red"
except requests.exceptions.RequestException as e:
latency = "N/A"
status_text = "❓ UNKNOWN"
message = f"Request error: {e}"
color = "grey"
except Exception as e:
latency = "N/A"
status_text = "❓ UNKNOWN"
message = f"Unexpected error: {e}"
color = "grey"
return {
"name": name,
"status": status_text,
"latency": f"{latency}ms" if isinstance(latency, (int, float)) else latency,
"message": message,
"color": color,
"url": url,
"last_checked": time.strftime("%H:%M:%S")
}
def create_dashboard_table(results: List[Dict[str, Any]], refresh_interval: float) -> Table:
"""
Creates a rich.Table object to display the API monitor results.
Args:
results: A list of dictionaries, each containing the result of an API check.
refresh_interval: The interval in seconds at which the dashboard refreshes.
Returns:
A rich.Table object ready for rendering.
"""
table = Table(
title=f"HTTP API Health Dashboard (Refresh every {refresh_interval}s)",
show_header=True,
header_style="bold magenta"
)
table.add_column("API Name", style="cyan", no_wrap=True)
table.add_column("URL", style="blue")
table.add_column("Status", style="bold", justify="center")
table.add_column("Latency", justify="right")
table.add_column("Message", style="dim")
table.add_column("Last Checked", style="dim", justify="right")
for result in results:
status_text = Text(result["status"], style=f"bold {result['color']}")
table.add_row(
result["name"],
result["url"],
status_text,
result["latency"],
result["message"],
result["last_checked"]
)
return table
# --- Main Execution Logic ---
def main():
"""
Main function to parse arguments, load monitors, and run the dashboard loop.
"""
parser = argparse.ArgumentParser(
description="HTTP API Health Monitor Dashboard.",
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument(
"--config",
"-c",
type=str,
default=DEFAULT_CONFIG_PATH,
help=f"Path to the YAML configuration file defining monitors.\n"
f"Defaults to '{DEFAULT_CONFIG_PATH}' or value of MONITOR_CONFIG_PATH env var."
)
parser.add_argument(
"--interval",
"-i",
type=float,
default=DEFAULT_REFRESH_INTERVAL,
help=f"Refresh interval for the dashboard in seconds.\n"
f"Defaults to {DEFAULT_REFRESH_INTERVAL}s or value of REFRESH_INTERVAL env var."
)
args = parser.parse_args()
console = Console()
monitor_config_path = args.config
refresh_interval = args.interval
try:
monitors = load_monitors(monitor_config_path)
console.print(f"[green]Successfully loaded {len(monitors)} monitors from '{monitor_config_path}'.[/green]")
except (FileNotFoundError, yaml.YAMLError, ValueError) as e:
console.print(f"[bold red]Error loading monitors: {e}[/bold red]", err=True)
return
# Initialize results list with basic info before first check
initial_results: List[Dict[str, Any]] = [
{
"name": m.get("name", "Unnamed API"),
"url": m.get("url", "N/A"),
"status": "⏳ PENDING",
"latency": "N/A",
"message": "Checking...",
"color": "blue",
"last_checked": "N/A"
} for m in monitors
]
# Use rich.Live to update the dashboard in place
with Live(
Panel(
Spinner("dots", text="[bold green]Initializing API Health Monitor...[/bold green]"),
border_style="green"
),
console=console,
screen=True, # Use the entire screen for the live display
refresh_per_second=4
) as live_display:
while True:
current_results: List[Dict[str, Any]] = []
for monitor in monitors:
result = check_api(monitor)
current_results.append(result)
dashboard_table = create_dashboard_table(current_results, refresh_interval)
live_display.update(dashboard_table)
time.sleep(refresh_interval)
if __name__ == "__main__":
main()
The script's behavior is primarily controlled by a YAML configuration file that defines the API endpoints to monitor, and optionally by environment variables or command-line arguments for general settings.
Environment Variables / Command-line Arguments:
MONITOR_CONFIG_PATH (Environment Variable) / --config or -c (Command-line Argument):
monitors.yaml in the current directory.export MONITOR_CONFIG_PATH=~/my_monitors.yaml or python3 api_monitor.py --config prod_apis.yamlREQUEST_TIMEOUT (Environment Variable):
5.0 seconds.export REQUEST_TIMEOUT=10.0REFRESH_INTERVAL (Environment Variable) / --interval or -i (Command-line Argument):
3.0 seconds.export REFRESH_INTERVAL=5.0 or python3 api_monitor.py -i 10monitors.yaml File Structure:
The core of the configuration is a YAML file, which should be a list of dictionaries. Each dictionary defines a single API endpoint to monitor with the following keys:
name (string, required): A human-readable name for the API endpoint.url (string, required): The full URL of the API endpoint to check.method (string, optional): The HTTP method to use (e.g., GET, POST, PUT).GETexpected_status (integer, optional): The HTTP status code expected for a successful response.200headers (dictionary, optional): A dictionary of HTTP headers to send with the request.headers: {'Content-Type': 'application/json', 'Authorization': 'Bearer YOUR_TOKEN'}body (string, optional): The request body to send. Useful for POST or PUT requests. For JSON bodies, ensure the Content-Type header is set to application/json.timeout (float, optional): Overrides the global REQUEST_TIMEOUT for this specific monitor.Example monitors.yaml:
- name: Google Homepage
url: https://www.google.com
method: GET
expected_status: 200
timeout: 3.0 # Shorter timeout for a quick check
- name: JSON Placeholder API (GET Post 1)
url: https://jsonplaceholder.typicode.com/posts/1
method: GET
expected_status: 200
- name: JSON Placeholder API (POST New Post)
url: https://jsonplaceholder.typicode.com/posts
method: POST
headers:
Content-Type: application/json
body: '{"title": "foo", "body": "bar", "userId": 1}'
expected_status: 201 # Expect 201 Created
- name: Non-Existent Endpoint (Expected Failure)
url: https://httpstat.us/500 # This endpoint always returns 500
method: GET
expected_status: 200 # This monitor will show as DOWN
- name: Intentional Timeout Test
url: https://httpstat.us/200?sleep=10000 # Sleeps for 10 seconds
method: GET
expected_status: 200
timeout: 2.0 # This monitor will show as TIMEOUT
First, ensure you have the required dependencies installed:
pip install requests rich PyYAML
Create a monitors.yaml file in the same directory as your api_monitor.py script, using the example structure provided in the Configuration section.
Basic Usage (with default monitors.yaml and default refresh interval):
python3 api_monitor.py
Expected Output: The terminal will clear, and a live-updating table will appear, similar to this:
HTTP API Health Dashboard (Refresh every 3.0s)
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┓
┃ API Name ┃ URL ┃ Status ┃ Latency ┃ Message ┃ Last Checked ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━┩
│ Google Homepage │ https://www.google.com │ ✅ UP │ 123.45ms │ Status 200 │ 10:30:05 │
│ JSON Placeholder API (GET... │ https://jsonplaceholder.typicode.com/posts/1 │ ✅ UP │ 78.90ms │ Status 200 │ 10:30:05 │
│ JSON Placeholder API (POST...│ https://jsonplaceholder.typicode.com/