Metadata-Version: 2.5
Name: fastapi-watch
Version: 1.8.0
Summary: Structured health and readiness check system for FastAPI
Project-URL: Repository, https://github.com/rgreen1207/fastapi-watch
License: MIT
Keywords: fastapi,health,kubernetes,liveness,probes,readiness
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Requires-Python: >=3.10
Requires-Dist: fastapi>=0.100.0
Requires-Dist: pydantic>=2.0
Provides-Extra: all
Requires-Dist: aio-pika>=9.4; extra == 'all'
Requires-Dist: aiohttp>=3.9; extra == 'all'
Requires-Dist: aiokafka>=0.9; extra == 'all'
Requires-Dist: aiomcache>=0.6; extra == 'all'
Requires-Dist: aiomysql>=0.2; extra == 'all'
Requires-Dist: asyncpg>=0.30; extra == 'all'
Requires-Dist: celery>=5.0; extra == 'all'
Requires-Dist: motor>=3.0; extra == 'all'
Requires-Dist: redis>=4.5.4; extra == 'all'
Requires-Dist: sqlalchemy>=2.0; extra == 'all'
Provides-Extra: celery
Requires-Dist: celery>=5.0; extra == 'celery'
Provides-Extra: http
Requires-Dist: aiohttp>=3.9; extra == 'http'
Provides-Extra: kafka
Requires-Dist: aiokafka>=0.9; extra == 'kafka'
Provides-Extra: memcached
Requires-Dist: aiomcache>=0.6; extra == 'memcached'
Provides-Extra: mongo
Requires-Dist: motor>=3.0; extra == 'mongo'
Provides-Extra: mysql
Requires-Dist: aiomysql>=0.2; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.30; extra == 'postgres'
Provides-Extra: rabbitmq
Requires-Dist: aio-pika>=9.4; extra == 'rabbitmq'
Requires-Dist: aiohttp>=3.9; extra == 'rabbitmq'
Provides-Extra: redis
Requires-Dist: redis>=4.5.4; extra == 'redis'
Provides-Extra: sqlalchemy
Requires-Dist: sqlalchemy>=2.0; extra == 'sqlalchemy'
Description-Content-Type: text/markdown

<h1 align="center">fastapi-watch</h1>
<p align="center">
  <em>Structured health and readiness checks for <a href="https://fastapi.tiangolo.com/">FastAPI</a>.</em>
</p>
<p align="center">
  <a href="https://github.com/rgreen1207/fastapi-watch/actions/workflows/publish.yml">
    <img src="https://github.com/rgreen1207/fastapi-watch/actions/workflows/publish.yml/badge.svg" alt="Test, Build & Publish">
  </a>
  <a href="https://pypi.org/project/fastapi-watch/">
    <img src="https://img.shields.io/pypi/v/fastapi-watch" alt="PyPI version">
  </a>
  <a href="https://pypi.org/project/fastapi-watch/">
    <img src="https://img.shields.io/pypi/pyversions/fastapi-watch" alt="Supported Python versions">
  </a>
  <a href="https://pepy.tech/projects/fastapi-watch">
    <img src="https://static.pepy.tech/personalized-badge/fastapi-watch?period=total&units=NONE&left_color=GREY&right_color=GREEN&left_text=downloads" alt="PyPI Downloads">
  </a>
</p>

---

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](https://github.com/rgreen1207/fastapi-watch/blob/main/DOCS.md)**.

---

## Installation

```bash
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

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

@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

# Create the app first: the registry mounts its routes on this instance.
app = FastAPI(lifespan=lifespan)
registry = HealthRegistry(app)

# Infrastructure probes observe your real calls: decorate them with @probe.watch
db_probe = PostgreSQLProbe(name="postgres")
cache_probe = RedisProbe(name="redis")
registry.add(db_probe)
registry.add(cache_probe, critical=False)

@db_probe.watch
async def fetch_user(user_id: int):
    return await pool.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
```

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

---

## Endpoints

| Endpoint | Purpose | Status |
|---|---|---|
| `GET /health/live` | Liveness — process is alive | `200` while the process runs (`401`/`403` without credentials when `auth=` is set) |
| `GET /health/ready` | Readiness — all critical probes passing | `200` / `503` / `404` for a `?probe=`/`?tag=` that selects nothing |
| `GET /health/status` | Full probe detail | `200` / `207` / `404` as above |
| `GET /health/dashboard` | Live HTML dashboard (SSE) | `200` |
| `GET /health/dashboard/stream-url` | Fresh dashboard stream URL and token — built-in dashboard with `auth=` only | `200` |
| `GET /health/metrics` | Prometheus text format 0.0.4 | `200` |
| `GET /health/history` | Rolling result history per probe | `200` / `503` when storage fails |
| `GET /health/alerts` | Probe state-change log | `200` / `503` when storage fails |
| `GET /health/startup` | Startup gate; 503 until `set_started()` | `200` / `503` |
| `GET /health/ready/stream` | SSE stream of readiness | stream / `404` as above |
| `GET /health/status/stream` | SSE stream of full probe detail (`?view=dashboard` trims per-route data to what the dashboard charts) | stream / `404` as above |
| `GET /health/maintenance` | Maintenance mode status | `200` |
| `POST /health/maintenance` | Enable maintenance mode — mounted only when `auth=` is set | `200` / `413` / `415` / `422` |
| `DELETE /health/maintenance` | Disable maintenance mode — mounted only when `auth=` is set | `200` |
| `GET /health/probes` | Registered probe config (no probe runs) | `200` |

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.

```python
@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`.

```python
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.

```python
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:

| Priority | Method | When to use |
|----------|--------|-------------|
| 1 — highest | `@probe.watch` | One route needs its own thresholds or name |
| 2 | `watch_router` | A whole router shares settings |
| 3 — lowest | `discover_routes` | Catch-all for everything not handled explicitly |

`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:

```bash
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):

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

A filter naming an unregistered probe or tag (a typo such as `?probe=postgress`), even alongside names that match, gets `404` with the unknown names, never a passing report, and the dashboard lists the name under "Requested but not registered". Known filters that together select nothing also get `404` (`"no probe matches every filter"`).

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

---

## Infrastructure probes

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

registry.add(PostgreSQLProbe(name="postgres"))
registry.add(RedisProbe(name="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

```python
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.

---

## Custom storage backend

Results, history, and alerts live in process memory by default. To share them across workers or keep them across restarts, use the bundled Redis backend (`pip install "fastapi-watch[redis]"`):

```python
import redis.asyncio as redis
from fastapi_watch import HealthRegistry, RedisProbeStorage

client = redis.from_url("redis://localhost:6379/0")  # decode_responses=True also works

registry = HealthRegistry(
    app,
    storage=RedisProbeStorage(client, prefix="myapp:health", max_entry_bytes=8192),
    history_size=120,            # entries kept per probe
    result_ttl_seconds=7200,     # per-entry age limit (0 = no age limit)
    max_alerts=120,
    alert_ttl_seconds=259200,
)
```

The four retention arguments are passed to `HealthRegistry` here, so the registry applies them to the backend through `configure()`. Leave all four out and the backend keeps its own constructor values (`RedisProbeStorage(..., max_results=..., result_ttl_seconds=..., alert_ttl_seconds=..., max_alerts=...)`, same defaults).

It is safe to point at the Redis your app already uses: it never runs `KEYS` or `SCAN`, every write is a single `MULTI/EXEC` round trip, and all keys sit under `prefix`. Cleanup of removed probes (`forget()`) only deletes names the same process wrote and no longer registers, so apps and replicas sharing a prefix never delete each other's data, even during a rolling deploy. History, alerts and the dashboard show only the app's own probes, so apps sharing a prefix do not see each other's data. Still, give each app its own prefix: two apps that register the same probe name on one prefix share that probe's data. Storage is capped by count, by age (per entry), and by bytes:

```
redis bytes ≤ probes × ((history_size + 1) × max_entry_bytes + probe name) + max_alerts × (~150 B + probe name)
```

Replicas sharing a prefix also share alert dedupe: `claim_alert(key, value, ttl_seconds)` is one atomic compare-and-set (a short Lua script), so only one replica sends each alert, a status the probe returns to within the window is sent again, and every claim key expires.

Latest results and history entries larger than `max_entry_bytes` drop their largest `details` values first and are marked `details["_truncated"] = true`. With the defaults, 20 probes take at most about 19 MB of history, and typically far less. The backend runs at most `max_connections` (default 4) Redis commands at once, so checking many probes together cannot drain a connection pool your app shares.

To write your own backend (PostgreSQL, Memcached, ...), implement the `ProbeStorage` protocol. See [DOCS.md — Custom storage backends](https://github.com/rgreen1207/fastapi-watch/blob/main/DOCS.md#custom-storage-backends).

---

## Security

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

```python
registry = HealthRegistry(app, auth={"type": "basic", "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):

```python
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:

```python
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](https://github.com/rgreen1207/fastapi-watch/blob/main/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.*
