This lesson walks the implementation, file by file. Five short modules, each under fifty lines, each doing one job. By the end you'll have a working api-monitor you can run against a real endpoint. Read top to bottom — every section builds on the previous one. If something doesn't compile, check that you've completed the file before it.
Step 1 — Data models (src/models.py)
Two dataclasses describe the shape of everything else: the inputs (Endpoint) and the outputs (CheckResult).
# src/models.py
from dataclasses import dataclass, field, asdict
from typing import Optional
@dataclass
class Endpoint:
name: str
url: str
expected_status: int = 200
max_time_ms: int = 2000
expected_fields: list = field(default_factory=list)
@dataclass
class CheckResult:
name: str
url: str
expected_status: int
actual_status: Optional[int]
actual_time_ms: Optional[int]
passed: bool
message: str
def to_dict(self) -> dict:
return asdict(self)Endpoint mirrors one row of the config JSON. CheckResult is what the monitor produces — the data the reporter then renders. actual_status and actual_time_ms are Optional[int] because both can be None if the call never reached the server (DNS failure, timeout). to_dict() is convenience for JSON output.
Two patterns from chapter 5 in play: field(default_factory=list) for the optional list (you can't write = [] in a dataclass), and asdict(self) for converting to plain data ready for json.dump.
Step 2 — Config loader (src/config_loader.py)
Read the JSON, validate the shape, return a list of Endpoints. Custom exceptions (chapter 6) make config problems distinct from check problems:
# src/config_loader.py
import json
from pathlib import Path
from typing import List
from .models import Endpoint
class ConfigError(Exception):
"""Raised when the config file is missing, malformed, or invalid."""
REQUIRED_FIELDS = ("name", "url")
def load(config_path: str) -> List[Endpoint]:
path = Path(config_path)
if not path.exists():
raise ConfigError(f"config file not found: {config_path}")
try:
data = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as e:
raise ConfigError(f"config is not valid JSON: {e}") from e
if not isinstance(data, dict) or "endpoints" not in data:
raise ConfigError("config must be an object with an 'endpoints' list")
endpoints = data["endpoints"]
if not isinstance(endpoints, list) or not endpoints:
raise ConfigError("'endpoints' must be a non-empty list")
result = []
for i, raw in enumerate(endpoints):
for f in REQUIRED_FIELDS:
if f not in raw:
raise ConfigError(f"endpoint #{i} missing required field {f!r}")
result.append(Endpoint(
name=raw["name"],
url=raw["url"],
expected_status=raw.get("expected_status", 200),
max_time_ms=raw.get("max_time_ms", 2000),
expected_fields=raw.get("expected_fields", []),
))
return resultA few details worth dwelling on:
from .models import Endpoint— relative import within thesrcpackage. The.says "the same package."- Two specific failure modes raised as
ConfigError. The caller canexcept ConfigErroronce and handle every config problem, regardless of cause. raise ConfigError(...) from e— chains the originalJSONDecodeErrorso the traceback shows both. Chapter 6'sfrom epattern.raw.get(key, default)for optionals;raw[key]for requireds. The defaults come fromEndpoint's field defaults — they're the contract.
Step 3 — Monitor (src/monitor.py)
The heart of the script. Two functions: check_endpoint does the work for one URL; check_all loops them with per-endpoint exception handling.
# src/monitor.py
import requests
from typing import List
from .models import Endpoint, CheckResult
def check_endpoint(endpoint: Endpoint) -> CheckResult:
"""Check one endpoint. Always returns a CheckResult — never raises."""
try:
response = requests.get(
endpoint.url,
timeout=max(endpoint.max_time_ms / 1000 * 2, 5), # generous outer cap
)
except requests.Timeout:
return _fail(endpoint, None, None, "timed out")
except requests.ConnectionError as e:
return _fail(endpoint, None, None, f"connection error: {e}")
except requests.RequestException as e:
return _fail(endpoint, None, None, f"request error: {e}")
actual_status = response.status_code
actual_time_ms = int(response.elapsed.total_seconds() * 1000)
if actual_status != endpoint.expected_status:
return _fail(endpoint, actual_status, actual_time_ms,
f"expected {endpoint.expected_status}, got {actual_status}")
if actual_time_ms > endpoint.max_time_ms:
return _fail(endpoint, actual_status, actual_time_ms,
f"slow: {actual_time_ms}ms > {endpoint.max_time_ms}ms")
if endpoint.expected_fields:
try:
body = response.json()
except ValueError:
return _fail(endpoint, actual_status, actual_time_ms,
"expected JSON body, got non-JSON")
if not isinstance(body, dict):
return _fail(endpoint, actual_status, actual_time_ms,
f"expected JSON object, got {type(body).__name__}")
missing = [f for f in endpoint.expected_fields if f not in body]
if missing:
return _fail(endpoint, actual_status, actual_time_ms,
f"missing fields: {', '.join(missing)}")
return CheckResult(
name=endpoint.name,
url=endpoint.url,
expected_status=endpoint.expected_status,
actual_status=actual_status,
actual_time_ms=actual_time_ms,
passed=True,
message="ok",
)
def _fail(endpoint: Endpoint, status, time_ms, msg) -> CheckResult:
return CheckResult(
name=endpoint.name,
url=endpoint.url,
expected_status=endpoint.expected_status,
actual_status=status,
actual_time_ms=time_ms,
passed=False,
message=msg,
)
def check_all(endpoints: List[Endpoint]) -> List[CheckResult]:
"""Run every check; one failure should not stop the others."""
return [check_endpoint(ep) for ep in endpoints]The shape of check_endpoint is guard-and-return. Each contract — network reachable, status matches, time within bound, JSON shape — gets a try or an if. Failure produces a CheckResult and returns immediately; success falls through to the bottom. Reads like prose; no nested else ladder.
_fail is a private helper (chapter 5's leading-underscore convention). It centralises the construction of a failed CheckResult so the body of check_endpoint stays focused on what failed rather than how to record it.
check_all is one line — a list comprehension over the endpoints. Each call to check_endpoint is independent; the comprehension keeps going if one returns a failure.
Step 4 — Reporter (src/reporter.py)
Two reports — console for humans, JSON for CI:
# src/reporter.py
import json
from pathlib import Path
from typing import List
from .models import CheckResult
GREEN = "\033[92m"
RED = "\033[91m"
RESET = "\033[0m"
def print_console(results: List[CheckResult]) -> None:
print(f"api-monitor — checking {len(results)} endpoints")
for r in results:
mark = f"{GREEN}✅{RESET}" if r.passed else f"{RED}❌{RESET}"
time_part = f"{r.actual_time_ms:>4}ms" if r.actual_time_ms is not None else " -- ms"
status_part = f"status={r.actual_status}" if r.actual_status is not None else "status= -- "
print(f"{mark} {r.name:<16} {status_part:<14} time={time_part} {r.message}")
passed = sum(1 for r in results if r.passed)
failed = len(results) - passed
print(f"\n{len(results)} checks: {passed} passed, {failed} failed")
def write_json(results: List[CheckResult], output_dir: str = "output") -> str:
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
path = out / "report.json"
payload = {
"total": len(results),
"passed": sum(1 for r in results if r.passed),
"failed": sum(1 for r in results if not r.passed),
"checks": [r.to_dict() for r in results],
}
with path.open("w", encoding="utf-8") as f:
json.dump(payload, f, indent=2)
return str(path)print_console uses ANSI colour codes — the \033[92m and \033[91m sequences make the ✅ green and the ❌ red on any modern terminal. Skip them for plain output if you prefer.
write_json produces a payload with totals at the top and the per-check details below. CI tools can read it directly; humans can cat it and skim. The out.mkdir(parents=True, exist_ok=True) line ensures the output directory exists before writing — chapter 4's idiom.
For an optional CSV report, add a write_csv function with csv.DictWriter (chapter 4) using [r.to_dict() for r in results] as the rows.
Step 5 — Entry point (main.py)
The thin CLI that wires everything together:
# main.py
import sys
import argparse
from src import config_loader
from src.config_loader import ConfigError
from src.monitor import check_all
from src.reporter import print_console, write_json
def main() -> int:
parser = argparse.ArgumentParser(description="API health-check monitor")
parser.add_argument("config", help="path to endpoints.json")
parser.add_argument("--output-dir", default="output",
help="where to write the JSON report (default: output/)")
args = parser.parse_args()
try:
endpoints = config_loader.load(args.config)
except ConfigError as e:
print(f"config error: {e}", file=sys.stderr)
return 2
results = check_all(endpoints)
print_console(results)
json_path = write_json(results, output_dir=args.output_dir)
print(f"report written to {json_path}")
return 0 if all(r.passed for r in results) else 1
if __name__ == "__main__":
sys.exit(main())Three things to notice:
argparse— Python's standard CLI parser. Readssys.argvfor you, generates--help, validates required arguments. A few lines unlocks a polished CLI.- Exit codes that mean something.
0for full success,1for any check failed (a normal "build red" signal),2for config errors (a setup problem distinct from a check failure). CI scripts can branch on these. if __name__ == "__main__":guard — chapter 6's idiom. Lets themainfunction be importable for pytest.
Running it
A real run, end to end:
python main.py config/endpoints.jsonOutput:
api-monitor — checking 3 endpoints
✅ health status=200 time= 87ms ok
✅ users-list status=200 time= 312ms ok
❌ admin-locked status=200 time= 45ms expected 401, got 200
3 checks: 2 passed, 1 failed
report written to output/report.json
Exit code is 1. CI sees the non-zero, fails the build, attaches output/report.json as an artefact. Job done.
The script in motion
Step 1 of 7
main.py kicks off
argparse parses the CLI args. The config path is required; --output-dir has a default.
Seven steps, five short files, one clean output. That's the whole capstone in motion.
Wiring it into CI
The script is now ready for a CI pipeline. A minimal GitHub Actions step:
- name: Smoke-check staging APIs
run: |
python -m venv venv
venv/bin/pip install -r requirements.txt
venv/bin/python main.py config/endpoints.json
- name: Upload report
if: always()
uses: actions/upload-artifact@v4
with:
name: api-monitor-report
path: output/report.jsonThe if: always() ensures the report is uploaded even when the script fails — exactly when you want it. The non-zero exit from main.py propagates up; the job turns red; downstream test jobs are skipped.
Where to next
You've built a real, useful Python tool. The final lesson reflects on what you've learnt and points at where to go next — and lists five stretch goals that turn this script into a portfolio-worthy piece of work.