boolean-healthcheck-sdk 0.1.0

Framework-agnostic health-check SDK to expose /health (liveness + readiness) from Python apps

boolean-healthcheck-sdk (Python) — Endpoint /health agnóstico

SDK de lado servidor para que cualquier app Python (Django, FastAPI, Flask, scripts) exponga un endpoint /health robusto que combine liveness ("estoy vivo": proceso responde, disco/RAM ok) y readiness ("puedo trabajar": DB, Redis, RabbitMQ, APIs de terceros).

Es agnóstico del framework: provee el motor de checks (HealthRegistry) y un handler puro (health_handler) que montás donde quieras.

Orquestador (k8s, load balancer, uptime monitor) ──▶ GET /health ──▶ HealthRegistry ──▶ [DB, Redis, RabbitMQ, HTTP, disco, RAM]

Instalación

pip install -e "./sdks/healthcheck-python"            # core (incluye HTTPCheck + disco)

# Extras opcionales por dependencia (sólo lo que tu app use):
pip install -e "./sdks/healthcheck-python[postgres]"  # psycopg
pip install -e "./sdks/healthcheck-python[redis]"     # redis
pip install -e "./sdks/healthcheck-python[rabbitmq]"  # pika
pip install -e "./sdks/healthcheck-python[system]"    # psutil (MemoryCheck)
pip install -e "./sdks/healthcheck-python[dev]"       # pytest + responses

Importar el paquete nunca importa los drivers opcionales. Sólo al instanciar un check (p.ej. RedisCheck(url=...)) sin el driver instalado se lanza MissingDriverError con el comando de instalación.

Concepto

• HealthRegistry — registra checks y los corre en paralelo con timeout (responde rápido y nunca cuelga). Cache TTL opcional para no golpear las dependencias en cada poll.
• HealthCheck — unidad de chequeo con name, critical y check() (devuelve detalle o lanza excepción). Un check critical caído → 503 DOWN; uno no-crítico caído → 200 DEGRADED.
• health_handler — función pura (headers, client_ip) -> (http_status, body). Público devuelve {"status": "UP"}; el detalle por dependencia sólo se expone con API key / IP autorizada.

Uso básico

from boolean_healthcheck_sdk import HealthRegistry, health_handler
from boolean_healthcheck_sdk.checks import (
    PostgresCheck, RedisCheck, RabbitMQCheck, HTTPCheck, DiskSpaceCheck,
)

registry = (
    HealthRegistry(timeout=2.0, cache_ttl=5.0, version="1.4.0")
    .register(DiskSpaceCheck("/", min_free_ratio=0.10))           # liveness
    .register(PostgresCheck(connection=db_conn))                  # SELECT 1
    .register(RedisCheck(client=redis_client, critical=False))    # degrada, no tumba
    .register(RabbitMQCheck(connection=amqp_conn))
    .register(HTTPCheck("https://pagos.proveedor.com/ping"))      # tercero crítico
)

# El handler es agnóstico: lo llamás desde cualquier framework.
handler = health_handler(registry, detail_api_key="super-secreto")

Los checks reciben conexiones/clientes ya existentes (no gestionan el pool). Como conveniencia también aceptan dsn=/url= para abrir una conexión efímera.

Montaje por framework (ejemplos, sin acoplamiento)

Django

# urls.py
from django.http import JsonResponse
from django.urls import path
from myapp.health import handler  # el health_handler de arriba

def health(request):
    status, body = handler(
        headers=request.headers,
        client_ip=request.META.get("REMOTE_ADDR"),
    )
    return JsonResponse(body, status=status)

urlpatterns = [path("health", health)]

FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/health")
def health(request: Request):
    status, body = handler(
        headers=request.headers,
        client_ip=request.client.host if request.client else None,
    )
    return JSONResponse(body, status_code=status)

Flask / WSGI plano

@app.get("/health")
def health():
    status, body = handler(headers=dict(request.headers), client_ip=request.remote_addr)
    return body, status

Códigos de estado

Situación	status	HTTP
Todos los checks OK	UP	200
Sólo checks no-críticos caídos	DEGRADED	200
Al menos un check crítico caído / timeout	DOWN	503

Respuesta

Pública (sin autorización):

{ "status": "UP" }

Detallada (con X-Health-Key válida o IP en allowed_ips):

{
  "status": "DEGRADED",
  "checks": [
    { "name": "postgres", "status": "UP", "latency_ms": 1.4, "critical": true },
    { "name": "redis", "status": "DEGRADED", "latency_ms": 2003.0, "critical": false,
      "detail": "timed out after 2.000s" }
  ],
  "version": "1.4.0",
  "uptime_seconds": 3601.2
}

Seguridad del detalle

handler = health_handler(
    registry,
    detail_api_key="super-secreto",      # se compara en tiempo constante
    detail_header="X-Health-Key",        # default
    allowed_ips=["10.0.0.0", "127.0.0.1"],
)

Sin API key correcta ni IP autorizada, el handler devuelve sólo {"status": ...} — no revela qué dependencia falló.

Checks built-in

Check	Dep (extra)	Probe
DiskSpaceCheck	— (stdlib)	shutil.disk_usage vs umbral de libre
MemoryCheck	psutil ([system])	% memoria usada vs umbral
PostgresCheck	psycopg ([postgres], sólo con dsn=)	SELECT 1
RedisCheck	redis ([redis], sólo con url=)	PING
RabbitMQCheck	pika ([rabbitmq], sólo con url=)	conexión abierta
HTTPCheck	requests (core)	GET a URL, status 2xx

Check propio: subclasá HealthCheck o usá registry.add("nombre", callable, critical=...).

Manejo de errores

Excepción	Cuándo ocurre
MissingDriverError	instanciás un check cuyo driver opcional no está instalado
HealthSDKError	clase base de todos los errores del SDK

Las fallas de las dependencias no lanzan excepción: se reflejan en el status/http_status del reporte.

Tests del SDK

cd sdks/healthcheck-python
pip install -e ".[dev]"
pytest

Usa responses para el HTTPCheck y clientes/conexiones falsos para DB/Redis (sin servicios reales).