Catch failed health checks before your users report them. In this tutorial, you’ll build an uptime monitor that checks websites and APIs, records results locally, and sends notifications when a service goes down or recovers.
The project runs on your laptop or a small server. SQLite keeps your monitoring history under your control, and a configurable webhook connects alerts to a self-hosted receiver or a service such as Slack.
What you’ll build
· HTTP checks with timeouts, configurable expected status codes, and bounded concurrency. · Confirmed outage and recovery states that filter brief failures. · Persistent check history and notification retries. · A bearer-protected status API. · A Docker Compose deployment with persistent storage.
This implementation suits a handful of services. A check validates the HTTP response status; it does not inspect page content or verify every dependency behind an application.
Prerequisites
· Python 3.12 and basic Python knowledge. · A terminal with curl; commands below use Bash. · Docker and Docker Compose for the container deployment. · A website or API to monitor. We’ll start with a local HTTP server.
Project structure
uptime-monitor/
├── app.py
├── receiver.py
├── requirements.txt
├── monitors.json
├── .env.example
├── .env # generated; keep private
├── .gitignore
├── .dockerignore
├── Dockerfile
├── compose.yaml
└── data/ # generated SQLite database and WAL files
- Create the project and install dependencies
mkdir uptime-monitor
cd uptime-monitor
python -m venv .venv
source .venv/bin/activate
Create requirements.txt:
fastapi==0.143.1
uvicorn==0.54.0
httpx==0.28.1
python-dotenv==1.2.4
Install the packages:
python -m pip install -r requirements.txt
FastAPI serves the status API, HTTPX performs asynchronous checks, and Python’s built-in sqlite3 module stores results.
- Configure secrets and monitored services
Create .env.example:
API_TOKEN=
CONFIG_PATH=monitors.json
DATABASE_PATH=data/uptime.db
CHECK_INTERVAL_SECONDS=30
CHECK_TIMEOUT_SECONDS=5
FAILURE_THRESHOLD=3
RECOVERY_THRESHOLD=2
CHECK_CONCURRENCY=5
HISTORY_DAYS=7
WEBHOOK_URL=
WEBHOOK_TOKEN=
Generate your local configuration and API token:
cp .env.example .env
python - <<'PY'
from pathlib import Path
import secrets
path = Path('.env')
path.write_text(path.read_text().replace(
'API_TOKEN=', 'API_TOKEN=' + secrets.token_hex(32), 1
))
PY
chmod 600 .env
Create .gitignore:
.env
.venv/
data/
__pycache__/
WEBHOOK_URL can stay empty: state changes will still appear in application logs. WEBHOOK_TOKEN, when supplied, becomes the receiver’s bearer token. Treat webhook URLs as secrets because many providers embed credentials in them.
Create monitors.json:
[
{
"id": "local-service",
"name": "Local test service",
"url": "http://127.0.0.1:9000/",
"expected_status": [200]
}
]
Add more objects to monitor more services. Each ID must be unique. For endpoints that legitimately return either status, use "expected_status": [200, 204].
Configuration loads at startup. Restart the monitor after editing it. Keep an ID stable for the same target; use a new ID when replacing a target with a different service so its historical state stays separate.
- Implement the monitor
Create app.py:
import asyncio
import json
import logging
import os
import secrets
import sqlite3
import time
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field, HttpUrl, TypeAdapter
load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logging.getLogger("httpx").setLevel(logging.WARNING)
log = logging.getLogger("uptime")
def positive_int(name, default):
value = int(os.getenv(name, str(default)))
if value < 1:
raise ValueError(f"{name} must be positive")
return value
INTERVAL = positive_int("CHECK_INTERVAL_SECONDS", 30)
TIMEOUT = positive_int("CHECK_TIMEOUT_SECONDS", 5)
FAILURES = positive_int("FAILURE_THRESHOLD", 3)
RECOVERIES = positive_int("RECOVERY_THRESHOLD", 2)
CONCURRENCY = positive_int("CHECK_CONCURRENCY", 5)
HISTORY_DAYS = positive_int("HISTORY_DAYS", 7)
DB_PATH = Path(os.getenv("DATABASE_PATH", "data/uptime.db"))
API_TOKEN = os.getenv("API_TOKEN", "")
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "")
WEBHOOK_TOKEN = os.getenv("WEBHOOK_TOKEN", "")
if len(API_TOKEN) < 32:
raise ValueError("Set API_TOKEN to a random token of at least 32 characters")
if WEBHOOK_URL:
WEBHOOK_URL = str(HttpUrl(WEBHOOK_URL))
class Monitor(BaseModel):
id: str = Field(pattern=r"^[a-z0-9_-]{1,64}$")
name: str = Field(min_length=1, max_length=100)
url: HttpUrl
expected_status: list[int] = Field(default_factory=lambda: [200], min_length=1)
monitors = TypeAdapter(list[Monitor]).validate_json(
Path(os.getenv("CONFIG_PATH", "monitors.json")).read_text()
)
if not monitors or len({m.id for m in monitors}) != len(monitors):
raise ValueError("Configure at least one monitor, with unique IDs")
for monitor in monitors:
if monitor.url.username or monitor.url.password:
raise ValueError("Use URLs without embedded credentials")
if any(code < 100 or code > 599 for code in monitor.expected_status):
raise ValueError("Expected status codes must be between 100 and 599")
def iso(timestamp):
return None if timestamp is None else datetime.fromtimestamp(
timestamp, timezone.utc
).isoformat(timespec="seconds")
def open_database():
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
db = sqlite3.connect(DB_PATH)
db.row_factory = sqlite3.Row
db.execute("PRAGMA journal_mode=WAL")
db.executescript("""
CREATE TABLE IF NOT EXISTS state (
monitor_id TEXT PRIMARY KEY, status TEXT NOT NULL DEFAULT 'unknown',
failures INTEGER NOT NULL DEFAULT 0,
successes INTEGER NOT NULL DEFAULT 0, changed_at REAL
);
CREATE TABLE IF NOT EXISTS checks (
id INTEGER PRIMARY KEY, monitor_id TEXT NOT NULL,
checked_at REAL NOT NULL, ok INTEGER NOT NULL,
status_code INTEGER, latency_ms INTEGER NOT NULL, detail TEXT
);
CREATE INDEX IF NOT EXISTS checks_monitor_time
ON checks(monitor_id, checked_at);
CREATE INDEX IF NOT EXISTS checks_time ON checks(checked_at);
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL,
payload TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
next_attempt REAL NOT NULL DEFAULT 0
);
""")
with db:
for monitor in monitors:
db.execute("INSERT OR IGNORE INTO state(monitor_id) VALUES (?)",
(monitor.id,))
return db
def record(db, monitor, ok, code, latency, detail):
now = time.time()
event = None
with db:
old = db.execute("SELECT * FROM state WHERE monitor_id=?",
(monitor.id,)).fetchone()
failures = 0 if ok else old["failures"] + 1
successes = old["successes"] + 1 if ok else 0
status = old["status"]
if ok and (status != "down" or successes >= RECOVERIES):
status = "up"
elif not ok and failures >= FAILURES:
status = "down"
changed = status != old["status"]
db.execute("""UPDATE state SET status=?, failures=?, successes=?,
changed_at=? WHERE monitor_id=?""",
(status, failures, successes,
now if changed else old["changed_at"], monitor.id))
db.execute("""INSERT INTO checks(monitor_id, checked_at, ok,
status_code, latency_ms, detail) VALUES (?, ?, ?, ?, ?, ?)""",
(monitor.id, now, int(ok), code, latency, detail))
if changed and (status == "down" or old["status"] == "down"):
event = {
"event_id": str(uuid.uuid4()),
"text": f"[{status.upper()}] {monitor.name}: {detail} at {iso(now)}",
}
if WEBHOOK_URL:
db.execute("INSERT INTO outbox(event_id, payload) VALUES (?, ?)",
(event["event_id"], json.dumps(event)))
if event:
log.info("%s", event["text"])
async def probe(db, client, semaphore, monitor):
async with semaphore:
start = time.monotonic()
code, ok = None, False
try:
async with asyncio.timeout(TIMEOUT):
async with client.stream("GET", str(monitor.url)) as response:
code = response.status_code
ok = code in monitor.expected_status
detail = f"HTTP {code}"
except (httpx.HTTPError, TimeoutError) as error:
ok = False
detail = type(error).__name__
latency = round((time.monotonic() - start) * 1000)
record(db, monitor, ok, code, latency, detail)
async def check_loop(db, client):
semaphore = asyncio.Semaphore(CONCURRENCY)
while True:
await asyncio.gather(*(probe(db, client, semaphore, m) for m in monitors))
with db:
db.execute("DELETE FROM checks WHERE checked_at < ?",
(time.time() - HISTORY_DAYS * 86400,))
await asyncio.sleep(INTERVAL)
async def notification_loop(db, client):
while True:
row = db.execute("SELECT * FROM outbox ORDER BY id LIMIT 1").fetchone()
if not WEBHOOK_URL or row is None or row["next_attempt"] > time.time():
await asyncio.sleep(1)
continue
headers = {"Idempotency-Key": row["event_id"]}
if WEBHOOK_TOKEN:
headers["Authorization"] = f"Bearer {WEBHOOK_TOKEN}"
try:
async with asyncio.timeout(TIMEOUT):
async with client.stream("POST", WEBHOOK_URL, headers=headers,
json=json.loads(row["payload"])) as response:
if not 200 <= response.status_code < 300:
raise RuntimeError(f"HTTP {response.status_code}")
except (httpx.HTTPError, TimeoutError, RuntimeError) as error:
attempts = row["attempts"] + 1
delay = min(300, 2 ** min(attempts, 9))
with db:
db.execute("UPDATE outbox SET attempts=?, next_attempt=? WHERE id=?",
(attempts, time.time() + delay, row["id"]))
log.warning("Notification retry in %ss: %s", delay, type(error).__name__)
else:
with db:
db.execute("DELETE FROM outbox WHERE id=?", (row["id"],))
log.info("Notification delivered: %s", row["event_id"])
def report_failure(task):
if not task.cancelled():
error = task.exception()
if error:
log.error("Background worker %s stopped", task.get_name(),
exc_info=(type(error), error, error.__traceback__))
@asynccontextmanager
async def lifespan(app):
db = open_database()
app.state.db = db
async with httpx.AsyncClient(timeout=TIMEOUT, follow_redirects=False,
headers={"User-Agent": "PSC-Uptime/1.0"}) as client:
tasks = [asyncio.create_task(check_loop(db, client), name="checks"),
asyncio.create_task(notification_loop(db, client), name="notifications")]
for task in tasks:
task.add_done_callback(report_failure)
app.state.tasks = tasks
try:
yield
finally:
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
db.close()
app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
@app.get("/healthz")
async def health():
if any(task.done() for task in app.state.tasks):
raise HTTPException(503, "A background worker stopped")
return {"ok": True}
@app.get("/api/status")
async def status(authorization: str | None = Header(default=None)):
if not secrets.compare_digest((authorization or "").encode(),
f"Bearer {API_TOKEN}".encode()):
raise HTTPException(401, "Invalid bearer token")
db = app.state.db
result = []
for monitor in monitors:
state = dict(db.execute("SELECT * FROM state WHERE monitor_id=?",
(monitor.id,)).fetchone())
latest = db.execute("""SELECT checked_at, ok, status_code, latency_ms, detail
FROM checks WHERE monitor_id=?
ORDER BY id DESC LIMIT 1""", (monitor.id,)).fetchone()
summary = db.execute("""SELECT COUNT(*) AS samples, AVG(ok)*100 AS percent
FROM checks WHERE monitor_id=? AND checked_at>=?""",
(monitor.id, time.time() - HISTORY_DAYS * 86400)).fetchone()
last_check = dict(latest) if latest else None
if last_check:
last_check["checked_at"] = iso(last_check["checked_at"])
last_check["ok"] = bool(last_check["ok"])
result.append({
"id": monitor.id, "name": monitor.name, "url": str(monitor.url),
"status": state["status"], "changed_at": iso(state["changed_at"]),
"failure_streak": state["failures"], "success_streak": state["successes"],
"last_check": last_check, "samples": summary["samples"],
"check_success_percent": round(summary["percent"], 2)
if summary["percent"] is not None else None,
})
pending = db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
return {"history_days": HISTORY_DAYS, "pending_notifications": pending,
"monitors": result}
The implementation has three parts:
- Probes request response headers, compare the status code, and record the result. Streaming avoids reading the full response body. asyncio.timeout() bounds the overall request; HTTPX’s own timeouts also protect individual network phases.
- State tracking changes up to down after consecutive failures and requires consecutive successes to recover. A first successful check establishes the initial up state silently. An initially unreachable service still generates a confirmed outage alert.
- Notification delivery sends the oldest queued event and retries failures independently of probing. The result, state update, and queued event share one SQLite transaction, so a confirmed transition cannot be committed without its notification being queued when webhooks are enabled.
One check cycle completes before the next interval begins. Actual sampling time includes the cycle’s duration, so three failures do not imply an exact 90-second detection time.
Run one Uvicorn worker and one application instance. Each process starts its own background loops; multiple workers or replicas would duplicate checks and notifications.
- Run it locally and test notifications
For a quicker test, change these values in .env:
CHECK_INTERVAL_SECONDS=5
CHECK_TIMEOUT_SECONDS=2
WEBHOOK_URL=http://127.0.0.1:9100/alerts
Start a service to monitor
In one terminal:
python -m http.server 9000 --bind 127.0.0.1
The server returns HTTP 200 at /, matching the configuration.
Start a local webhook receiver
Create receiver.py:
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
class Receiver(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get("Content-Length", "0"))
event = json.loads(self.rfile.read(length))
print(json.dumps(event, indent=2), flush=True)
self.send_response(204)
self.end_headers()
HTTPServer(("127.0.0.1", 9100), Receiver).serve_forever()
In another terminal:
python receiver.py
This receiver prints alerts and acknowledges them with HTTP 204. It is a local testing utility; use your notification provider or an authenticated receiver for deployment.
Start the monitor
In another terminal, activate the project’s virtual environment and run:
source .venv/bin/activate
uvicorn app:app --host 127.0.0.1 --port 8000 --workers 1
Inspect the status API
In a terminal with the virtual environment activated:
export API_TOKEN="$(python -c 'from dotenv import dotenv_values; print(dotenv_values(".env")["API_TOKEN"])')"
curl -fsS http://127.0.0.1:8000/healthz
curl -fsS http://127.0.0.1:8000/api/status \
-H "Authorization: Bearer $API_TOKEN" | python -m json.tool
After the first successful check, the service’s status should be up. The response includes its latest check, latency, confirmation counters, successful-check percentage, and pending notification count.
Confirm that authentication is enforced:
curl -i http://127.0.0.1:8000/api/status
Expect HTTP 401.
Trigger an outage and recovery
- Stop the HTTP server with Ctrl+C.
- Wait for three failed checks. The receiver should print one [DOWN] event.
- Leave the server stopped for another few checks. The monitor should keep recording failures without generating more outage events.
- Restart the HTTP server. After two successful checks, the receiver should print one [UP] event.
During confirmation, the reported status retains its previous state. Check last_check.ok and the counters to distinguish the latest observation from the confirmed state.
Verify retries and persistence
Stop the receiver, then cause another outage and recovery. /api/status should show queued notifications while checks continue.
Restart the monitor while the receiver remains stopped. The pending events and service state should survive. Start the receiver again; the queued outage and recovery arrive in that order after their retry delays.
Delivery is at least once: if a receiver accepts an alert and the monitor crashes before deleting it, the alert can be sent again. Every retry retains the same event_id and Idempotency-Key header. Deduplication requires support from the receiver; the test receiver simply prints duplicates.
- Connect a notification provider
Set WEBHOOK_URL to a reachable HTTPS endpoint that accepts JSON shaped like this:
{
"event_id": "a-stable-uuid-for-this-transition",
"text": "[DOWN] Website: HTTP 503 at 2026-10-14T12:00:00+00:00"
}
The receiver must return a 2xx response to acknowledge delivery. For Slack, use an incoming webhook URL and leave WEBHOOK_TOKEN empty; its message comes from text. Providers with other payload formats, such as Discord’s content field, need an adapter or a change to the notification payload.
For a self-hosted receiver, set WEBHOOK_TOKEN if it expects bearer authentication. Notifications contain the monitor’s name, result, and timestamp, while full check history remains in your SQLite database.
Restart the application after changing these settings. Enabling a webhook sends future state transitions; it does not generate a snapshot alert for a service already marked down.
- Package it with Docker Compose
Create Dockerfile:
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt \
&& useradd --uid 10001 --create-home uptime \
&& mkdir /app/data \
&& chown uptime:uptime /app/data
COPY --chmod=644 app.py .
USER uptime
EXPOSE 8000
CMD ["uvicorn", "app:app",