philiprehberger-health-check

Health check endpoint builder for web applications.

Installation

pip install philiprehberger-health-check

Usage

from philiprehberger_health_check import HealthCheck, checks

hc = HealthCheck()
hc.add("database", lambda: True)
hc.add(*checks.tcp("redis", "localhost", 6379))
hc.add(*checks.disk_space("disk", "/", min_free_gb=2))

result = hc.run()
print(result.status)  # "healthy" or "unhealthy"

for check in result.checks:
    print(f"{check.name}: {'OK' if check.healthy else check.message}")

Check Dependencies

from philiprehberger_health_check import HealthCheck

hc = HealthCheck()
hc.add("database", check_db)
hc.add("migrations", check_migrations, depends_on=["database"])

result = hc.run()
# If "database" fails, "migrations" is skipped automatically

Per-Check Timeouts

from philiprehberger_health_check import HealthCheck

hc = HealthCheck(timeout=10.0)  # global timeout
hc.add("fast-db", check_db, timeout=2.0)  # override for this check
hc.add("slow-report", generate_report, timeout=30.0)

result = hc.run()
# Each check uses its own timeout; others fall back to the global 10s

Remediation Actions

from philiprehberger_health_check import CheckResult, HealthCheck

def restart_cache(result: CheckResult) -> None:
    print(f"Restarting cache after failure: {result.message}")

hc = HealthCheck()
hc.add("cache", check_cache, on_failure=restart_cache)

result = hc.run()
# If "cache" fails, restart_cache is called automatically

Check History and Metrics

from philiprehberger_health_check import HealthCheck

hc = HealthCheck(history_size=50)
hc.add("database", check_db)

hc.run()
hc.run()
hc.run()

history = hc.history("database")  # list of past CheckResult objects
rate = hc.success_rate("database")  # float between 0.0 and 1.0

Async Execution

import asyncio
from philiprehberger_health_check import HealthCheck

hc = HealthCheck()
hc.add("database", check_db)
hc.add("cache", check_cache)
hc.add("migrations", check_migrations, depends_on=["database"])

result = await hc.run_async()
# Independent checks run concurrently; dependencies are respected

Serializing for HTTP

from philiprehberger_health_check import HealthCheck

hc = HealthCheck()
hc.add("database", lambda: True)

# Convert a HealthResult into a JSON-serializable dict
result = hc.run()
body = result.to_dict()
# {"status": "healthy", "uptime_seconds": 0.01, "checks": [...]}

# Or get a (status_code, body_dict) tuple ready for any web framework
status, body = hc.to_response()
# status == 200 when healthy, 503 when unhealthy

# Override the status codes if you need to
status, body = hc.to_response(ok_status=204, fail_status=500)

Built-in Checks

from philiprehberger_health_check import HealthCheck, checks

hc = HealthCheck()

# TCP connectivity
hc.add(*checks.tcp("postgres", "db.example.com", 5432, timeout=3))

# Disk space
hc.add(*checks.disk_space("storage", "/data", min_free_gb=5))

# Memory usage (Linux only, skipped on other platforms)
hc.add(*checks.memory("ram", max_percent=85))

# Custom check
hc.add(*checks.custom("cache", lambda: cache.ping()))

API

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT