Metadata-Version: 2.4
Name: healthcheckx
Version: 0.1.1
Summary: Production-grade health checks for Python applications
Author-email: Soumen Samanta <you@example.com>
Project-URL: Homepage, https://github.com/Soumen3/healthcheckx
Project-URL: Repository, https://github.com/Soumen3/healthcheckx
Project-URL: Issues, https://github.com/Soumen3/healthcheckx/issues
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Topic :: System :: Monitoring
Classifier: Topic :: Internet :: WWW/HTTP
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == "redis"
Provides-Extra: rabbitmq
Requires-Dist: pika>=1.3; extra == "rabbitmq"
Provides-Extra: postgresql
Requires-Dist: psycopg2-binary>=2.9; extra == "postgresql"
Provides-Extra: mysql
Requires-Dist: mysql-connector-python>=8.0; extra == "mysql"
Provides-Extra: oracle
Requires-Dist: oracledb>=3.4; extra == "oracle"
Provides-Extra: mssql
Requires-Dist: pymssql>=2.2; extra == "mssql"
Provides-Extra: mongodb
Requires-Dist: pymongo>=4.0; extra == "mongodb"
Provides-Extra: all
Requires-Dist: redis>=5.0; extra == "all"
Requires-Dist: pika>=1.3; extra == "all"
Requires-Dist: psycopg2-binary>=2.9; extra == "all"
Requires-Dist: mysql-connector-python>=8.0; extra == "all"
Requires-Dist: oracledb>=3.4; extra == "all"
Requires-Dist: pymssql>=2.2; extra == "all"
Requires-Dist: pymongo>=4.0; extra == "all"
Dynamic: license-file

# healthcheckx

[![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**healthcheckx** is a production-grade health check library for Python applications. It provides a simple, extensible framework for monitoring the health of various services and components in your application.

## 🚀 Features

- **🔌 Framework Agnostic**: Core library works independently, with optional adapters for FastAPI, Flask, and Django
- **🎯 Built-in Health Checks**: Pre-built checks for popular services (Redis, RabbitMQ, PostgreSQL, MySQL, SQLite, Oracle, MS SQL Server, MongoDB)
- **⚡ Simple & Extensible**: Easy to add custom health checks with a clean API
- **🎭 Graceful Degradation**: Distinguishes between healthy, degraded, and unhealthy states
- **⏱️ Performance Tracking**: Measures execution time for each health check
- **🛡️ Error Handling**: Catches exceptions gracefully and returns meaningful status
- **🔧 Configurable Timeouts**: Non-blocking checks with configurable timeouts
- **📦 Optional Dependencies**: Install only what you need

## 📦 Installation

### Basic Installation

```bash
pip install healthcheckx
```

### With Optional Dependencies

Install support for specific services:

```bash
# Redis support
pip install healthcheckx[redis]

# RabbitMQ support
pip install healthcheckx[rabbitmq]

# PostgreSQL support
pip install healthcheckx[postgresql]

# MySQL support
pip install healthcheckx[mysql]

# SQLite (built-in, no extra dependencies needed)

# Oracle support
pip install healthcheckx[oracle]

# MS SQL Server support
pip install healthcheckx[mssql]

# MongoDB support
pip install healthcheckx[mongodb]

# Install everything
pip install healthcheckx[all]
```

## 🎯 Quick Start

### Basic Usage

```python
from healthcheckx import Health

# Create health check instance
health = Health()

# Register built-in checks (method chaining supported)
health.redis_check("redis://localhost:6379") \
      .postgresql_check("postgresql://user:pass@localhost/db") \
      .mongodb_check("mongodb://localhost:27017")

# Run all checks
results = health.run()

# Inspect results
for result in results:
    print(f"{result.name}: {result.status} ({result.duration_ms:.2f}ms)")
    if result.message:
        print(f"  Message: {result.message}")
```

### Aggregate Status

```python
from healthcheckx import Health, overall_status

health = Health()
health.redis_check("redis://localhost:6379") \
      .postgresql_check("postgresql://user:pass@localhost/db")

results = health.run()
status = overall_status(results)  # Returns: "healthy", "degraded", or "unhealthy"

print(f"Overall Status: {status}")
```

## 🔧 Built-in Health Checks

### Cache Systems

#### Redis

```python
health.redis_check("redis://localhost:6379", timeout=2)
```

### Message Queues

#### RabbitMQ

```python
health.rabbitmq_check("amqp://guest:guest@localhost:5672", timeout=2)
```

### Relational Databases

#### PostgreSQL

```python
health.postgresql_check("postgresql://user:password@localhost:5432/mydb", timeout=3)
```

#### MySQL

```python
health.mysql_check("mysql://root:password@localhost:3306/mydb", timeout=3)
```

#### SQLite

```python
# File-based database
health.sqlite_check("/path/to/database.db", timeout=3)

# In-memory database
health.sqlite_check(":memory:")
```

#### Oracle

```python
# URL format
health.oracle_check("oracle://user:password@localhost:1521/XEPDB1", timeout=3)

# TNS format
health.oracle_check("user/password@localhost:1521/service_name")
```

#### MS SQL Server

```python
health.mssql_check("mssql://sa:Password@localhost:1433/master", timeout=3)
```

### NoSQL Databases

#### MongoDB

```python
# Local MongoDB
health.mongodb_check("mongodb://localhost:27017", timeout=3)

# With authentication
health.mongodb_check("mongodb://user:password@localhost:27017/mydb")

# MongoDB Atlas (cloud)
health.mongodb_check("mongodb+srv://user:pass@cluster.mongodb.net/db")
```

## 🌐 Framework Integration

### FastAPI

```python
from fastapi import FastAPI
from healthcheckx import Health, FastAPIAdapter

app = FastAPI()
health = Health()

# Register health checks
health.redis_check("redis://localhost:6379") \
      .postgresql_check("postgresql://user:pass@localhost/db")

# Add health endpoint
adapter = FastAPIAdapter(health)
app.get("/health")(adapter.endpoint)

# Returns:
# - HTTP 200 for healthy/degraded
# - HTTP 503 for unhealthy
# - JSON body with status and individual check results
```

**Response Example:**
```json
{
  "status": "healthy",
  "checks": [
    {
      "name": "redis",
      "status": "healthy",
      "duration_ms": 12.5
    },
    {
      "name": "postgresql",
      "status": "healthy",
      "duration_ms": 45.3
    }
  ]
}
```

### Flask

```python
from flask import Flask
from healthcheckx import Health, flask_health_endpoint

app = Flask(__name__)
health = Health()

# Register health checks
health.redis_check("redis://localhost:6379") \
      .mysql_check("mysql://root:pass@localhost:3306/db")

# Add health endpoint
app.route("/health")(flask_health_endpoint(health))
```

### Django

```python
# urls.py
from django.urls import path
from healthcheckx import Health, django_health_view

health = Health()
health.redis_check("redis://localhost:6379") \
      .postgresql_check("postgresql://user:pass@localhost/db")

urlpatterns = [
    path('health/', django_health_view(health)),
]
```

## 🎨 Custom Health Checks

### Creating Custom Checks

You can create custom health checks by defining a function that returns a `CheckResult`:

```python
from healthcheckx import Health, CheckResult, HealthStatus

def custom_api_check():
    """Check external API availability"""
    try:
        import requests
        response = requests.get("https://api.example.com/status", timeout=2)
        
        if response.status_code == 200:
            return CheckResult("external-api", HealthStatus.healthy)
        else:
            return CheckResult(
                "external-api", 
                HealthStatus.degraded,
                f"API returned {response.status_code}"
            )
    except Exception as e:
        return CheckResult(
            "external-api",
            HealthStatus.unhealthy,
            str(e)
        )

# Register custom check
health = Health()
health.register(custom_api_check)
results = health.run()
```

### Reusable Custom Checks (Factory Pattern)

```python
def create_disk_check(path: str, min_free_percent: float = 10.0):
    """Factory function to create disk space check"""
    def check():
        import shutil
        stat = shutil.disk_usage(path)
        free_percent = (stat.free / stat.total) * 100
        
        if free_percent >= min_free_percent:
            return CheckResult("disk", HealthStatus.healthy)
        elif free_percent >= min_free_percent / 2:
            return CheckResult(
                "disk",
                HealthStatus.degraded,
                f"Low disk space: {free_percent:.1f}% free"
            )
        else:
            return CheckResult(
                "disk",
                HealthStatus.unhealthy,
                f"Critical disk space: {free_percent:.1f}% free"
            )
    
    return check

# Use the custom check
health = Health()
health.register(create_disk_check("/", min_free_percent=15.0))
results = health.run()
```

### HTTP Service Check Example

```python
def create_http_check(url: str, timeout: int = 3):
    """Check HTTP endpoint availability"""
    def check():
        try:
            import requests
            response = requests.get(url, timeout=timeout)
            
            if response.status_code == 200:
                return CheckResult("http-check", HealthStatus.healthy)
            else:
                return CheckResult(
                    "http-check",
                    HealthStatus.unhealthy,
                    f"HTTP {response.status_code}"
                )
        except Exception as e:
            return CheckResult("http-check", HealthStatus.unhealthy, str(e))
    
    return check

health = Health()
health.register(create_http_check("https://api.example.com/ping"))
```

## 📊 Health Status Levels

healthcheckx supports three levels of health status:

1. **`healthy`**: Service is functioning normally
2. **`degraded`**: Service is operational but impaired (e.g., high latency, partial functionality)
3. **`unhealthy`**: Service is down or failing

### Aggregation Rules

When determining overall status from multiple checks:

- If **any** check is `unhealthy` → Overall status is `unhealthy`
- Else if **any** check is `degraded` → Overall status is `degraded`
- Else → Overall status is `healthy`

## ⚙️ API Reference

### `Health` Class

#### Methods

- **`register(check: Callable) -> Health`**: Register a health check function
- **`run() -> List[CheckResult]`**: Execute all registered checks and return results
- **`redis_check(redis_url: str, timeout: int = 2) -> Health`**: Register Redis check
- **`rabbitmq_check(amqp_url: str, timeout: int = 2) -> Health`**: Register RabbitMQ check
- **`postgresql_check(dsn: str, timeout: int = 3) -> Health`**: Register PostgreSQL check
- **`mysql_check(dsn: str, timeout: int = 3) -> Health`**: Register MySQL check
- **`sqlite_check(db_path: str, timeout: int = 3) -> Health`**: Register SQLite check
- **`oracle_check(dsn: str, timeout: int = 3) -> Health`**: Register Oracle check
- **`mssql_check(dsn: str, timeout: int = 3) -> Health`**: Register MS SQL Server check
- **`mongodb_check(connection_string: str, timeout: int = 3) -> Health`**: Register MongoDB check

### `CheckResult` Class

```python
@dataclass
class CheckResult:
    name: str                    # Name/identifier of the check
    status: HealthStatus         # Health status (healthy/degraded/unhealthy)
    message: str | None = None   # Optional message (usually for errors)
    duration_ms: float | None = None  # Execution time in milliseconds
```

### `HealthStatus` Enum

```python
class HealthStatus(str, Enum):
    healthy = "healthy"
    degraded = "degraded"
    unhealthy = "unhealthy"
```

### `overall_status()` Function

```python
def overall_status(results: List[CheckResult]) -> HealthStatus:
    """Determine overall health status from individual check results"""
```

## 🛠️ Advanced Usage

### Conditional Health Checks

```python
import os
from healthcheckx import Health

health = Health()

# Always check database
health.postgresql_check(os.getenv("DATABASE_URL"))

# Only check Redis in production
if os.getenv("ENV") == "production":
    health.redis_check(os.getenv("REDIS_URL"))
```

### Dynamic Check Registration

```python
databases = [
    {"type": "postgresql", "dsn": "postgresql://localhost/db1"},
    {"type": "mysql", "dsn": "mysql://localhost/db2"},
]

health = Health()
for db in databases:
    if db["type"] == "postgresql":
        health.postgresql_check(db["dsn"])
    elif db["type"] == "mysql":
        health.mysql_check(db["dsn"])
```

### Custom Timeout Configuration

```python
health = Health()

# Different timeouts for different services
health.redis_check("redis://localhost:6379", timeout=1)  # Fast check
health.postgresql_check("postgresql://localhost/db", timeout=5)  # Slower check
health.mongodb_check("mongodb://localhost:27017", timeout=3)  # Medium check
```

### Health Check with Environment Variables

```python
import os
from healthcheckx import Health

health = Health()

# Load from environment
if redis_url := os.getenv("REDIS_URL"):
    health.redis_check(redis_url)

if db_url := os.getenv("DATABASE_URL"):
    health.postgresql_check(db_url)

if mongodb_url := os.getenv("MONGODB_URL"):
    health.mongodb_check(mongodb_url)
```

## 🧪 Testing Your Application

```python
from healthcheckx import Health, HealthStatus

def test_application_health():
    health = Health()
    health.postgresql_check("postgresql://localhost/test_db")
    health.redis_check("redis://localhost:6379")
    
    results = health.run()
    
    # Assert all checks passed
    for result in results:
        assert result.status == HealthStatus.healthy
        assert result.duration_ms < 1000  # All checks under 1 second
```

## 🐳 Docker Health Checks

Use healthcheckx in Docker health checks:

```dockerfile
FROM python:3.11
WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

# Health check endpoint
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```

## 🔍 Monitoring & Observability

Integrate with monitoring tools:

```python
from healthcheckx import Health, HealthStatus
import logging

health = Health()
health.redis_check("redis://localhost:6379") \
      .postgresql_check("postgresql://localhost/db")

results = health.run()

# Log results
for result in results:
    if result.status == HealthStatus.unhealthy:
        logging.error(f"{result.name} is unhealthy: {result.message}")
    elif result.status == HealthStatus.degraded:
        logging.warning(f"{result.name} is degraded: {result.message}")

# Send to metrics system (Prometheus, etc.)
for result in results:
    # metrics.gauge(f"health.{result.name}.duration_ms", result.duration_ms)
    # metrics.gauge(f"health.{result.name}.status", 1 if result.status == "healthy" else 0)
    pass
```

## 📝 Best Practices

1. **Set Appropriate Timeouts**: Keep health check timeouts short (2-5 seconds) to avoid blocking
2. **Separate Readiness from Liveness**: Use different health check endpoints for Kubernetes readiness/liveness probes
3. **Cache Results**: For high-traffic endpoints, consider caching health check results for a few seconds
4. **Monitor Check Duration**: Track `duration_ms` to identify slow dependencies
5. **Use Graceful Degradation**: Return `degraded` status when service is operational but impaired
6. **Avoid Heavy Operations**: Health checks should be lightweight (simple ping/select operations)
7. **Handle Exceptions**: Let healthcheckx handle exceptions gracefully by returning appropriate status

## 🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## 📄 License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

## 🔗 Links

- **Homepage**: [https://github.com/Soumen3/healthcheckx](https://github.com/Soumen3/healthcheckx)
- **Issues**: [https://github.com/Soumen3/healthcheckx/issues](https://github.com/Soumen3/healthcheckx/issues)
- **PyPI**: [https://pypi.org/project/healthcheckx/](https://pypi.org/project/healthcheckx/)

## 👨‍💻 Author

**Soumen Samanta**

---

Made with ❤️ for the Python community
