Metadata-Version: 2.4
Name: healthcheckx
Version: 0.1.2
Summary: Production-grade health checks for Python applications
Author-email: Soumen Samanta <you@example.com>
License: MIT License
        
        Copyright (c) 2026 Soumen Samanta
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
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
