fastapi-watch

Structured health and readiness checks for FastAPI.

Add /health/* endpoints to any FastAPI app in minutes. Probes observe real traffic — no synthetic requests — and stream live results to a built-in dashboard and Prometheus endpoint.

For full documentation see DOCS.md.

Installation

pip install fastapi-watch

# With service-specific extras
pip install "fastapi-watch[postgres,redis]"
pip install "fastapi-watch[all]"

zsh users: quote extras to avoid glob expansion: pip install "fastapi-watch[redis]"

Available extras: postgres, mysql, sqlalchemy, redis, memcached, rabbitmq, kafka, mongo, celery

Quick start

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_watch import HealthRegistry
from fastapi_watch.probes import PostgreSQLProbe, RedisProbe

app = FastAPI()
registry = HealthRegistry(app)

# Infrastructure probes
registry.add(PostgreSQLProbe(url="postgresql://user:pass@localhost/mydb"))
registry.add(RedisProbe(url="redis://localhost:6379"), critical=False)

@asynccontextmanager
async def lifespan(app):
    # Automatically monitor every route — no decorators needed
    registry.discover_routes(tags=["api"], max_error_rate=0.05)
    registry.set_started()
    yield

app = FastAPI(lifespan=lifespan)

Every route is now monitored with real-traffic data. Health endpoints are live at /health/*.

Endpoints

The prefix defaults to /health and is configurable: HealthRegistry(app, prefix="/ops/health").

Route monitoring

The fastest way to monitor your API is auto-discovery. fastapi-watch monitors real traffic — no synthetic polling, no wasted requests.

discover_routes — monitor everything at once

One call after all routers are included. Every route gets a passive probe with no decorators required.

@asynccontextmanager
async def lifespan(app):
    registry.discover_routes(
        tags=["api"],              # tag all probes for filtering
        max_error_rate=0.05,       # alert if error rate exceeds 5%
        include_paths=["/api/*"],  # whitelist — only monitor these routes
        exclude_paths=["/api/admin"],  # exclude even if include_paths matches
    )
    registry.set_started()
    yield

Auto-discovered probes use GET /items/{id} style descriptions so they're immediately recognizable in the dashboard.

watch_router — monitor a specific router

Scope monitoring to one router with its own tags, thresholds, and criticality. Call it after app.include_router.

app.include_router(users_router, prefix="/users")
app.include_router(orders_router, prefix="/orders")

registry.watch_router(users_router, tags=["users"], max_error_rate=0.01)
registry.watch_router(orders_router, tags=["orders"], critical=False)

Then filter health checks by router: GET /health/ready?tag=users

@probe.watch — full control on one route

For routes that need custom thresholds, a specific name, or tight SLOs.

from fastapi_watch import FastAPIRouteProbe

checkout_probe = FastAPIRouteProbe(
    name="checkout",
    description="Payment processing",
    tags=["payments"],
    max_error_rate=0.001,
    slow_call_threshold_ms=200,
)

@app.post("/checkout")
@checkout_probe.watch
async def checkout():
    ...

registry.add(checkout_probe)

Priority system

When all three approaches are used together, each route is monitored exactly once — the highest-priority wins:

discover_routes and watch_router skip any route already covered by a higher-priority method — no conflicts, no double counting.

Tag-based filtering

All probes accept tags=[...]. FastAPI route tags (@app.get("/items", tags=["store"])) are automatically merged in. Filter any endpoint by tag:

GET /health/ready?tag=payments         # only payment probes
GET /health/status?tag=users,orders    # users OR orders probes
GET /health/status/stream?tag=payments # filtered live stream

Filter by exact probe name with ?probe= (comma-separated, AND'd with ?tag= when both are given):

GET /health/status?probe=postgresql,redis   # only these two probes
GET /health/status?tag=payments&probe=checkout  # must match both

The dashboard shows tag chips on each probe card and a clickable filter bar to isolate groups at a glance.

Infrastructure probes

from fastapi_watch.probes import PostgreSQLProbe, RedisProbe, HttpProbe, TCPProbe

registry.add(PostgreSQLProbe(url="postgresql://..."))
registry.add(RedisProbe(url="redis://..."), critical=False)
registry.add(HttpProbe(name="payments-api"), critical=True)
registry.add(TCPProbe(host="kafka.internal", port=9092))

Passive probes observe real calls — use @probe.watch on any function to track its latency, error rate, and throughput without making synthetic requests.

Alerting

from fastapi_watch.alerts import SlackAlerter, PagerDutyAlerter, OpsGenieAlerter

registry = HealthRegistry(
    app,
    alerters=[
        SlackAlerter(webhook_url="https://hooks.slack.com/..."),
        PagerDutyAlerter(routing_key="your-routing-key"),
        OpsGenieAlerter(api_key="your-api-key"),
    ],
)

Webhook URLs are validated at construction time — private/loopback/link-local IP targets are rejected to prevent SSRF.

Security

Health endpoints are publicly accessible by default — set auth in production:

registry = HealthRegistry(app, auth={"username": "ops", "password": "secret"})

To manually reset a circuit breaker (e.g. after you've fixed the underlying issue and don't want to wait for the cooldown):

registry.reset_circuit("postgresql")  # clears open state and failure count; returns self

To skip all route registration entirely and expose health status on your own endpoint with custom auth:

registry = HealthRegistry(app, serve_routes=False)

@app.get("/internal/health", dependencies=[Depends(my_auth)])
async def my_health():
    report = await registry.get_report()
    return report.model_dump()

See DOCS.md — Security for auth callables, SSRF protection on webhook alerters, probe error message handling, and probe name restrictions.

License

MIT

Claude used to write README, code annotation, help with test case coverage, and clean up my messy thoughts into readable code.