Health Checks

AMSDAL Server exposes health check endpoints for monitoring and orchestration platforms (Kubernetes, Docker, load balancers, etc.).

Endpoints

Liveness Probe

GET /api/probes/liveness/

Returns 200 OK if the server is running and all health checks pass, or 503 Service Unavailable if any check fails.

Response:

{
    "status": "success",
    "details": [
        {
            "status": "success",
            "service": "ConnectionsHealthchecker",
            "message": "Connection is alive"
        }
    ]
}

Readiness Probe

GET /api/probes/readiness/

Returns 200 OK when the server is ready to accept traffic.

---

Built-in Health Checkers

ConnectionsHealthchecker

Checks that all configured database connections are established and alive.

from amsdal_server.apps.healthcheck.services.checkers.connections import ConnectionsHealthchecker

This checker verifies:

• Each connection's is_connected status
• Each connection's is_alive status

If any connection is down, the health check returns an error status.

---

Custom Health Checkers

Create a custom health checker by subclassing BaseHealthchecker:

from amsdal_server.apps.healthcheck.services.checkers.base import BaseHealthchecker
from amsdal_server.apps.healthcheck.serializers import HealthcheckServiceResult, StatusEnum

class RedisHealthchecker(BaseHealthchecker):
    async def check(self) -> HealthcheckServiceResult:
        try:
            # Check your service
            await redis_client.ping()
            return HealthcheckServiceResult(
                status=StatusEnum.success,
                service=self.__class__.__name__,
                message='Redis is available',
            )
        except Exception as e:
            return HealthcheckServiceResult(
                status=StatusEnum.error,
                service=self.__class__.__name__,
                message=f'Redis check failed: {e}',
            )

Registering custom checkers

Add your checker to the HealthcheckService via the add_condition() method:

from amsdal_server.apps.healthcheck.services.healthcheck import HealthcheckService

healthcheck_service = HealthcheckService(conditions=[
    ConnectionsHealthchecker(),
    RedisHealthchecker(),
])

Or add it dynamically:

healthcheck_service.add_condition(RedisHealthchecker())

---

Response Format

Field	Type	Description
status	"success" | "error"	Overall status
details	list	Per-checker results
details[].status	"success" | "error"	Individual checker status
details[].service	str	Checker class name
details[].message	str	Human-readable status message

The overall status is "error" if any individual checker returns an error.

---

Kubernetes Example

livenessProbe:
  httpGet:
    path: /api/probes/liveness/
    port: 8000
  initialDelaySeconds: 10
  periodSeconds: 30

readinessProbe:
  httpGet:
    path: /api/probes/readiness/
    port: 8000
  initialDelaySeconds: 5
  periodSeconds: 10