Metadata-Version: 2.4
Name: firebreak-vitals
Version: 1.1.1
Summary: Structured healthcheck endpoints with shallow and deep response tiers, concurrent check execution, and built-in support for PostgreSQL, Redis, SQLAlchemy, and FastAPI.
Project-URL: Homepage, https://github.com/firebreak-io/vitals
Project-URL: Repository, https://github.com/firebreak-io/vitals
Project-URL: Issues, https://github.com/firebreak-io/vitals/issues
Author-email: Firebreak <oss@firebreak.io>
License-Expression: MIT
Keywords: async,asyncpg,fastapi,healthcheck,monitoring,redis,sqlalchemy
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: all
Requires-Dist: asyncpg>=0.29; extra == 'all'
Requires-Dist: fastapi>=0.110; extra == 'all'
Requires-Dist: httpx>=0.27; extra == 'all'
Requires-Dist: redis>=5.0; extra == 'all'
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == 'all'
Provides-Extra: asyncpg
Requires-Dist: asyncpg>=0.29; extra == 'asyncpg'
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == 'fastapi'
Requires-Dist: httpx>=0.27; extra == 'fastapi'
Provides-Extra: http
Requires-Dist: httpx>=0.27; extra == 'http'
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == 'redis'
Provides-Extra: sqlalchemy
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == 'sqlalchemy'
Description-Content-Type: text/markdown

# Vitals (Python)

Structured deep healthcheck endpoints for Python services. Register health checks, run them concurrently with timeouts, and expose them via FastAPI.

## Install

```bash
pip install vitals
```

Install with optional dependencies for the checks and integrations you need:

```bash
pip install vitals[asyncpg]      # PostgreSQL via asyncpg
pip install vitals[redis]        # Redis via redis-py
pip install vitals[sqlalchemy]   # PostgreSQL via SQLAlchemy
pip install vitals[fastapi]      # FastAPI integration
pip install vitals[all]          # Everything
```

## Quick Start

```python
from vitals import HealthcheckRegistry, Status, CheckResult
from vitals.checks.postgres import AsyncpgPoolCheck
from vitals.checks.redis import RedisPoolCheck
from vitals.integrations.fastapi import create_healthcheck_route
from fastapi import FastAPI

registry = HealthcheckRegistry(default_timeout=5.0)

# Add checks using existing connection pools
registry.add("postgres", AsyncpgPoolCheck(pool))
registry.add("redis", RedisPoolCheck(redis_pool))

# Or add a custom check
@registry.check("api")
async def check_api() -> CheckResult:
    start = time.monotonic()
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/ping")
    latency = (time.monotonic() - start) * 1000
    return CheckResult(
        status=Status.HEALTHY if resp.is_success else Status.OUTAGE,
        latency_ms=latency,
        message="" if resp.is_success else f"HTTP {resp.status_code}",
    )

# Mount as a FastAPI route
app = FastAPI()
app.include_router(create_healthcheck_route(
    registry,
    token=os.environ.get("HEALTHCHECK_TOKEN"),
))
```

## API

### `HealthcheckRegistry`

```python
registry = HealthcheckRegistry(default_timeout=5.0)
```

**`registry.add(name, check, *, timeout=None)`**

Register a named async check function.

```python
async def my_check() -> CheckResult:
    return CheckResult(status=Status.HEALTHY, latency_ms=0.0, message="")

registry.add("service", my_check, timeout=3.0)
```

**`registry.check(name, *, timeout=None)`**

Decorator form of `add`.

```python
@registry.check("service", timeout=3.0)
async def check_service() -> CheckResult:
    ...
```

**`await registry.run()`**

Execute all registered checks concurrently using `asyncio.TaskGroup`. Returns a `HealthcheckResponse` with the worst overall status.

```python
response = await registry.run()
response.status       # Status.HEALTHY
response.to_dict()    # {'status': 'healthy', 'timestamp': '...', 'checks': {...}}
response.http_status_code  # 200
```

### Status

```python
from vitals import Status

Status.HEALTHY   # 2
Status.DEGRADED  # 1
Status.OUTAGE    # 0

Status.from_string("degraded")  # Status.DEGRADED
Status.HEALTHY.json_value       # "healthy"
```

### Built-in Checks

#### PostgreSQL

```python
from vitals.checks.postgres import AsyncpgCheck, AsyncpgPoolCheck, SQLAlchemyCheck

# Fresh asyncpg connection each time
registry.add("pg", AsyncpgCheck("postgresql://localhost:5432/mydb"))

# Existing asyncpg pool
registry.add("pg", AsyncpgPoolCheck(pool))

# SQLAlchemy AsyncEngine
registry.add("pg", SQLAlchemyCheck(engine))
```

#### Redis

```python
from vitals.checks.redis import RedisCheck, RedisPoolCheck

# Fresh connection each time
registry.add("redis", RedisCheck("redis://localhost:6379"))

# Existing redis.asyncio connection pool
registry.add("redis", RedisPoolCheck(pool))
```

### Sync Checks

Wrap synchronous functions using `asyncio.to_thread`:

```python
from vitals import sync_check, Status, CheckResult

def check_disk() -> CheckResult:
    free = shutil.disk_usage("/").free
    return CheckResult(
        status=Status.HEALTHY if free > 1_000_000_000 else Status.DEGRADED,
        latency_ms=0.0,
        message=f"{free} bytes free",
    )

registry.add("disk", sync_check(check_disk))
```

### FastAPI Integration

```python
from vitals.integrations.fastapi import create_healthcheck_route

router = create_healthcheck_route(
    registry,
    token="my-secret-token",       # optional — omit to disable auth
    path="/healthcheck/deep",      # default
    query_param_name="token",      # default
    include_in_schema=False,       # default — hides from OpenAPI docs
)
app.include_router(router)
```

When a token is configured, requests must provide it via:
- Query parameter: `?token=my-secret-token`
- Bearer header: `Authorization: Bearer my-secret-token`

### Liveness Probes (`/ping`)

Never point a liveness probe at the healthcheck endpoint. The registry runs on
every request, so a transient DB / Redis / downstream blip fails the probe and,
after the orchestrator's failure threshold, restarts the container — turning a
brief dependency hiccup into cascading restarts.

Use a separate registry-free liveness endpoint that only confirms the process is
up. It never touches the registry, always returns HTTP `200`, and responds with
`{"status": "ok", "timestamp": ..., **metadata}`.

- liveness probe → `/ping` (registry-free, always 200)
- readiness / dependency checks → `/healthcheck/deep` (runs the registry)

```python
from vitals.integrations.fastapi import create_healthcheck_route, create_liveness_route

app.include_router(create_liveness_route(path="/ping"))          # registry-free, no auth
app.include_router(create_healthcheck_route(registry))           # runs the registry
```

`create_liveness_route` takes `path="/ping"`, optional `metadata`, and
`include_in_schema=False` (defaults). It is public (no token).

Framework-agnostic handler:

```python
from vitals import create_liveness_handler

ping = create_liveness_handler(metadata={"build": "stg-45d76e5"})
result = ping({})
# HandlerResult(status=200, body={"status": "ok", "timestamp": "...", "build": "stg-45d76e5"})
```

`metadata` follows the same reserved-key rules as the healthcheck handler
(`status`, `timestamp`, `checks`, `cachedAt` are rejected).

### Authentication

```python
from vitals import verify_token, extract_token

# Timing-safe token comparison
verify_token("provided-token", "expected-token")  # bool

# Extract token from request data
token = extract_token(
    query_params={"token": "abc"},
    authorization_header="Bearer abc",
    query_param_name="token",
)
```

## Response Format

```json
{
  "status": "healthy",
  "timestamp": "2025-02-26T12:00:00.000Z",
  "checks": {
    "postgres": {
      "status": "healthy",
      "latencyMs": 4.2,
      "message": ""
    },
    "redis": {
      "status": "healthy",
      "latencyMs": 1.1,
      "message": ""
    }
  }
}
```

| Overall Status | HTTP Code |
|---------------|-----------|
| `healthy` | `200` |
| `degraded` | `503` |
| `outage` | `503` |
| Auth failure | `403` |

## Requirements

- Python >= 3.11
