philiprehberger-health-check 0.4.0

Health check endpoint builder for web applications.

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

Function / Class	Description
HealthCheck(timeout=30.0, history_size=100)	Create a health check builder with optional global timeout and history size
hc.add(name, fn, depends_on=None, timeout=None, on_failure=None)	Register a check with optional dependencies, per-check timeout, and failure callback
hc.run()	Run all checks sequentially and return HealthResult
hc.run_async()	Run checks concurrently, respecting dependency order
hc.history(name)	Return list of past CheckResult objects for a check
hc.success_rate(name)	Return success rate as a float between 0.0 and 1.0
hc.to_response(ok_status=200, fail_status=503)	Run checks and return (http_status, body_dict) for web frameworks
checks.tcp(name, host, port, timeout=2)	TCP connectivity check
checks.disk_space(name, path, min_free_gb=1)	Disk free space check
checks.memory(name, max_percent=90)	Memory usage check (Linux)
checks.custom(name, fn)	Wrap any callable as a check
CheckResult	Dataclass: name, healthy, message, duration_ms
HealthResult	Dataclass: status, checks, uptime_seconds
HealthResult.to_dict()	Return a JSON-serializable dict suitable for an HTTP response

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