firebreak-vitals 1.1.0

Structured healthcheck endpoints with shallow and deep response tiers, concurrent check execution, and built-in support for PostgreSQL, Redis, SQLAlchemy, and FastAPI.

Vitals (Python)

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

Install

pip install vitals

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

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

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

registry = HealthcheckRegistry(default_timeout=5.0)

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

Register a named async check function.

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.

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

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

Status

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

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

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:

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

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

Authentication

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

{
  "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