python_outbox_core.health_check

Health check interface for outbox monitoring.

Best Practices Applied: 1. Standard health check contract 2. Multi-level status (healthy/degraded/unhealthy) 3. Detailed diagnostics for debugging 4. Prometheus-compatible metrics

References: - Health Check API: https://microservices.io/patterns/observability/health-check-api.html

Classes

HealthStatus	Health status levels.
OutboxHealthCheck	Health check for outbox worker.

Module Contents

class python_outbox_core.health_check.HealthStatus

Bases: str, enum.Enum

Health status levels.

HEALTHY = 'healthy'

DEGRADED = 'degraded'

UNHEALTHY = 'unhealthy'

class python_outbox_core.health_check.OutboxHealthCheck

Bases: abc.ABC

Health check for outbox worker.

Projects implement this to expose health endpoints. Useful for K8s liveness/readiness probes.

abstract check_health() → Dict[str, Any]

Async:

Perform health check. Returns dict with status, timestamp, checks.

async check_database() → Dict[str, Any]

Check if database is reachable.

async check_broker() → Dict[str, Any]

Check if message broker is reachable.

async check_outbox_lag() → Dict[str, Any]

Check outbox processing lag.

Returns unhealthy if too many pending events.