kuzbara 0.4.1
Fresh, organic health checks for Python.
Requires Python: >=3.9

# Kuzbara (كُزْبَرَة) 🌿

[![PyPI version](https://img.shields.io/pypi/v/kuzbara.svg)](https://pypi.org/project/kuzbara/)
[![Python versions](https://img.shields.io/pypi/pyversions/kuzbara.svg)](https://pypi.org/project/kuzbara/)
[![License](https://img.shields.io/pypi/l/kuzbara.svg)](https://github.com/suaue/kuzbara/blob/master/LICENSE)

**Fresh, organic health checks for Python.**

Kuzbara (Arabic for Coriander) is a robust, standard-compliant health check library. It solves the biggest pain point in modern Python services: **Running async and sync checks together without blocking your app.**

It automatically detects your database drivers (SQLAlchemy, asyncpg, psycopg2, redis-py) and executes them in the correct mode (Native Async or Threaded Sync).

## Features

*   **Hybrid Runner**: Safely runs blocking I/O (Disk, Psycopg2) alongside async tasks (Asyncpg, Redis) in the same request.
*   **Universal Probes**: One `PostgresProbe` class handles asyncpg, psycopg, and SQLAlchemy automatically.
*   **IETF Compliant**: Returns `application/health+json` (RFC Draft).
*   **Zero-Dependency Core**: The base library has 0 external requirements.
*   **Bulkhead Pattern**: Isolate health checks from your main application threads.
*   **CLI for CI/CD**: Run health checks from the command line with `kuzbara check`.
*   **HTML Dashboard**: Visual debugging via `?format=html` with status badges and latency bars.

## Timeout Behavior

Kuzbara uses different timeout strategies for async and sync probes:

### Async Probes (Hard Timeout)
- Timeout is **strictly enforced** via `asyncio.timeout()`
- If a probe exceeds its timeout, it is **cancelled immediately**
- Result: `status: "fail"` with message "Timeout after Xs"
- Use case: ASGI frameworks (FastAPI, Starlette)

### Sync Probes (Soft Timeout)
- Timeout is a **warning threshold**, not a hard limit
- The probe **always runs to completion** (Python cannot safely kill threads)
- If duration exceeds timeout, result is `status: "warn"` (not fail)
- For hard enforcement, configure driver-level timeouts (connection timeout, query timeout)
- Use case: WSGI frameworks (Flask, Django)

### Global Timeout
- Applies to the entire health check suite
- Hard enforced in both async and sync contexts
- If exceeded, all pending probes fail and request returns immediately

### Example Configuration
```python
from kuzbara import HealthCheck
from kuzbara.probes import HttpProbe

# Per-probe timeout (5 seconds)
probe = HttpProbe(url="https://api.example.com", timeout=5.0)

# Global timeout (30 seconds for all probes combined)
health = HealthCheck(global_timeout=30.0)
```

### Why Different Behaviors?
- **Async**: Event loop allows safe task cancellation
- **Sync**: Thread termination in Python is unsafe and can cause resource leaks
- **Recommendation**: For sync probes, always set driver-level timeouts lower than probe timeout

## Installation

```bash
pip install kuzbara
```

Optional extras (mix and match):

```bash
pip install "kuzbara[http]"      # For checking external APIs (httpx)
pip install "kuzbara[system]"    # For checking Memory usage (psutil)
pip install "kuzbara[redis]"     # For Redis probes
pip install "kuzbara[postgres]"  # For PostgreSQL probes (asyncpg)

# Or install everything
pip install "kuzbara[all]"
```

## Quick Start

### 1. The Registry

Define your checks in a central file (e.g., `health.py`).

```python
from kuzbara import HealthCheck
from kuzbara.probes import PostgresProbe, RedisProbe, DiskProbe

# Create the registry
health = HealthCheck(name="payment-service", version="1.0.0")

# --- Database Check ---
# You just pass the connection/pool. Kuzbara figures out the driver.
# Works with: asyncpg, psycopg2, psycopg3, SQLAlchemy (Sync & Async)
health.add_probe(PostgresProbe(
    conn=db_pool, 
    name="main-db"
))

# --- Redis Check ---
# Works with: redis-py (Sync) and redis.asyncio (Async)
health.add_probe(RedisProbe(
    client=redis_client, 
    name="cache-layer"
))

# --- System Check ---
# Runs in a thread automatically to avoid blocking the event loop
health.add_probe(DiskProbe(warning_mb=500))  # Uses platform default path
```

### 2. FastAPI

```python
from fastapi import FastAPI
from kuzbara.adapters.fastapi import create_health_router, create_health_lifespan
from .health import health  # Import your registry

# Use lifespan for automatic cleanup (FastAPI 0.100+)
app = FastAPI(lifespan=create_health_lifespan(health))

# Adds the /health endpoint
app.include_router(create_health_router(health))
```

### 3. Flask

```python
from flask import Flask
from kuzbara.adapters.flask import FlaskHealth
from .health import health  # Import your registry

app = Flask(__name__)

# Adds the /health endpoint (handles shutdown via atexit)
FlaskHealth(app, health_check=health)
```

### 4. Django

```python
# urls.py
from django.urls import path
from kuzbara.adapters.django import DjangoHealthView
from .health import health  # Import your registry

urlpatterns = [
    path('health/', DjangoHealthView.as_view(health_check=health)),
]
```

For proper cleanup, register the shutdown handler in your `AppConfig`:

```python
# apps.py
from django.apps import AppConfig


class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        from kuzbara.adapters.django import register_shutdown
        from .health import health
        register_shutdown(health)
```

### The Output

Hitting `GET /health` returns a structured IETF JSON response.

```json
{
  "status": "warn",
  "version": "1.0.0",
  "checks": {
    "main-db": [
      {
        "status": "pass",
        "componentType": "datastore",
        "observedValue": 1,
        "time": "2026-01-01T12:00:00+00:00",
        "metadata": {"latency_ms": 2.34},
        "critical": true
      }
    ],
    "disk": [
      {
        "status": "warn",
        "componentType": "system",
        "output": "Disk space low: 450MB < 500MB",
        "observedValue": 450,
        "time": "2026-01-01T12:00:00+00:00",
        "metadata": {"latency_ms": 0.12},
        "critical": true
      }
    ]
  },
  "metrics": {
    "duration_ms": 2.5,
    "probe_count": 2,
    "global_timeout_ms": 30000.0
  }
}
```

**Status Codes:**
- `200 OK` - status is `pass` or `warn`
- `503 Service Unavailable` - status is `fail`

### HTML Debug View

For visual debugging, add `?format=html` to any health endpoint:

```
GET /health?format=html
```

Or use the `Accept: text/html` header. This renders an interactive dashboard with:
- Status badges (green/yellow/red)
- Latency bars
- Expandable probe details

### Framework Adapters Reference

| Framework | Main Function | Creates | Auto-Shutdown |
|:----------|:--------------|:--------|:--------------|
| **FastAPI** | `create_health_router()` | `APIRouter` | Via `create_health_lifespan()` |
| **Flask** | `create_health_blueprint()` | `Blueprint` | Via `atexit` |
| **Django** | `DjangoHealthView.as_view()` | CBV | Via `register_shutdown()` |

> **Note:** Each adapter uses framework-native patterns (Router, Blueprint, CBV) for seamless integration.

## Available Probes

| Probe | Description | Install |
|:------|:------------|:--------|
| `DiskProbe` | Checks free disk space | *None (stdlib)* |
| `MemoryProbe` | Checks RAM usage | `kuzbara[system]` |
| `HttpProbe` | Checks external APIs | `kuzbara[http]` |
| `RedisProbe` | Checks Redis (Sync/Async) | `kuzbara[redis]` |
| `PostgresProbe` | Universal SQL check | `kuzbara[postgres]` |

### PostgresProbe Auto-Detection

Supports: `asyncpg`, `psycopg2`, `psycopg3`, `SQLAlchemy` (sync & async)

```python
from kuzbara.probes import PostgresProbe, SQLDriverType

# Auto-detection (works for standard library objects)
probe = PostgresProbe(conn=asyncpg_pool, name="db")

# Explicit driver (for wrapped connections, proxies, or mocks)
probe = PostgresProbe(
    conn=my_custom_pool,
    name="db",
    driver=SQLDriverType.PSYCOPG_SYNC,
)
```

## Configuration

```python
health = HealthCheck(
    name="payment-api",       # Service name
    version="abc123",         # Version (git hash, semver, etc.)
    global_timeout=30.0,      # Fail entire check if > 30s
    max_workers=4,            # Dedicated thread pool (bulkhead pattern)
    cache_ttl=5.0,            # Cache results for 5 seconds (anti-DDoS)
)
```

### Result Caching (Anti-DDoS)

To prevent aggressive polling from overwhelming your health check endpoints, use the `cache_ttl` parameter:

```python
health = HealthCheck(
    name="api",
    cache_ttl=5.0,  # Return cached results for 5 seconds
)
```

When a request arrives within the TTL window, the cached result is returned immediately without running the probes again. This is especially useful for Kubernetes liveness/readiness probes that poll frequently.

When the cache expires, Kuzbara uses a lock-based approach: the first request acquires the lock and runs the probes while other concurrent requests wait for the result. This prevents a "thundering herd" of concurrent probe executions.

Note: caching is handled internally by Kuzbara when serving requests (via the adapters). If you call the registry directly, use `HealthCheck.try_get_or_run()` / `HealthCheck.async_try_get_or_run()` rather than trying to manage cache state yourself.

### Critical vs Optional Probes

By default, all probes are critical—a failure causes the overall status to be `fail` (HTTP 503). For non-essential checks, set `critical=False`:

```python
from kuzbara.probes import RedisProbe, DiskProbe

# Critical: failure = global fail
health.add_probe(RedisProbe(client=redis, name="cache"))

# Non-critical: failure is recorded but doesn't affect global status
health.add_probe(DiskProbe(
    name="logs-disk",
    critical=False,  # Won't cause HTTP 503 if disk is full
))
```

The individual check still shows `"status": "fail"` in the JSON response, but the global status remains `pass` (HTTP 200) if only non-critical probes fail.

### Bulkhead Pattern

By setting `max_workers`, health checks run in an isolated thread pool:

```python
health = HealthCheck(max_workers=2)  # Isolate from main thread pool
```

### Hybrid Runner (Mixing Async and Sync Probes)

Kuzbara's runner automatically handles both async and sync probes in any environment:

- **ASGI apps (FastAPI)**: Async probes run natively on the event loop. Sync probes are offloaded to threads via `run_in_executor`.
- **WSGI apps (Flask/Django)**: Sync probes run in threads. All async probes are batched into a single thread that runs an ephemeral event loop with `asyncio.gather`.

This "hybrid runner" design means you can freely mix `asyncpg` (async), `psycopg2` (sync), and `redis.asyncio` (async) probes in the same health check—Kuzbara figures out the optimal execution strategy.

> **Note:** In WSGI mode, all async probes share one event loop thread. Since health checks are I/O-bound (waiting on network/disk), this has no practical throughput penalty—probes still run concurrently via asyncio.

### Probe Groups (Kubernetes Liveness vs Readiness)

Kubernetes distinguishes between liveness ("restart if dead") and readiness ("stop traffic if not ready") checks. Pointing both at the same endpoint is dangerous—if your database goes down, you don't want Kubernetes restarting your pods endlessly.

Use probe groups to separate them:

```python
from kuzbara import HealthCheck
from kuzbara.probes import DiskProbe, PostgresProbe

health = HealthCheck()

# Liveness: lightweight check (is the process alive?)
health.add_probe(DiskProbe(name="disk"), groups=["liveness", "readiness"])

# Readiness: full dependency check (can I serve traffic?)
health.add_probe(PostgresProbe(conn=db_pool, name="db"), groups=["readiness"])
```

Then enable separate endpoints in your adapter:

```python
# FastAPI
app.include_router(create_health_router(health, include_live_ready=True))

# Flask
FlaskHealth(app, health_check=health, include_live_ready=True)

# Django
urlpatterns = get_health_urlpatterns(health, include_live_ready=True)
```

This creates:
- `GET /health` → All probes
- `GET /health/live` → Only probes in `["liveness"]` group
- `GET /health/ready` → Only probes in `["readiness"]` group

Probes without explicit groups run on ALL endpoints (they're assigned to a special `__all__` group).

### Retry Logic

For flaky network connections, probes can retry on failure:

```python
from kuzbara.probes import HttpProbe

health.add_probe(HttpProbe(
    name="external-api",
    url="https://api.example.com/health",
    retry_attempts=2,   # Override default (if any)
    retry_delay=0.3,    # Wait 300ms between retries
))

# Global Defaults (Applied if probe doesn't specify)
health = HealthCheck(
    default_retry_attempts=3,
    default_retry_delay=1.0,  # 1s between attempts for all probes
)
```

Retry behavior:
- Only failures (exceptions, timeouts) trigger retries
- `WarnCondition` is **not** retried (it's intentional)
- Total latency in metadata includes all retry attempts

### Redacting Sensitive Data

For production environments, you may want to hide version and latency information:

```python
health = HealthCheck(
    redact_sensitive=True,  # Hides version and latency_ms from responses
)
```

This removes the `version` field and `metadata.latency_ms` from health check responses, preventing potential attackers from fingerprinting your application or measuring internal timing.

## Writing Custom Probes

### Decorator API (recommended for simple checks)

The fastest way to create a probe is with the `@health.probe()` decorator:

```python
from kuzbara import HealthCheck, WarnCondition

health = HealthCheck()

@health.probe(name="config-check", timeout=1.0)
def check_config_loaded():
    if not config.IS_LOADED:
        raise ValueError("Config not loaded!")
    return {"env": "production"}

# Async functions work too:
@health.probe(name="db-ping", component_type="datastore")
async def check_database():
    return await db.ping()
```

### Class-based API (for complex or reusable probes)

For probes that need constructor parameters or complex logic, subclass `BaseProbe`:

```python
from kuzbara import BaseProbe, WarnCondition

class QueueDepthProbe(BaseProbe):
    def __init__(self, queue):
        super().__init__(name="queue-depth", timeout=5.0)
        self.queue = queue

    async def check(self):
        depth = await self.queue.get_depth()
        if depth > 1000:
            raise WarnCondition(f"Queue backlog: {depth} items")
        return depth

# Sync probes work too - runner handles threading automatically
class MySyncProbe(BaseProbe):
    def check(self):
        # Blocking I/O is fine
        return "OK"
```

## Examples

See the [`examples/`](examples/) folder for runnable demos:

- **basic.py** — CLI usage without a web framework
- **custom_probe.py** — Writing custom probes (class-based and decorator-based)
- **fastapi_app.py** — FastAPI integration with HTML dashboard
- **flask_app.py** — Flask integration with HTML dashboard
- **django_demo.py** — Django integration

## CLI (CI/CD Integration)

Run health checks directly from the command line without starting a web server. Perfect for pre-deployment verification in CI/CD pipelines.

### Basic Usage

```bash
# Run all probes
kuzbara check myapp.health:health

# Override timeout
kuzbara check myapp.health:health --timeout 10

# Output formats
kuzbara check myapp.health:health --format text   # Human-readable (default)
kuzbara check myapp.health:health --format json   # IETF JSON
kuzbara check myapp.health:health --format html > report.html  # Save as file
```

### Exit Codes

| Code | Meaning |
|:-----|:--------|
| `0` | All probes passed (status is `pass`) |
| `1` | Some probes failed or warned (status is `fail` or `warn`) |
| `2` | Error loading or running health checks |

### Example CI/CD Usage

```yaml
# GitHub Actions
- name: Verify dependencies
  run: |
    pip install kuzbara
    kuzbara check myapp.health:health --timeout 30
```

## License

[MIT](LICENSE)