43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Mapping
|
|
from datetime import UTC, datetime
|
|
from typing import cast
|
|
|
|
from django.http import JsonResponse
|
|
from django.views.decorators.cache import never_cache
|
|
|
|
from apps.health.checks import check_backup, check_cache, check_celery, check_db
|
|
|
|
CRITICAL_CHECKS = {"db", "cache"}
|
|
|
|
|
|
@never_cache
|
|
def health_view(request):
|
|
checks: dict[str, Mapping[str, object]] = {
|
|
"db": check_db(),
|
|
"cache": check_cache(),
|
|
"celery": check_celery(),
|
|
"backup": check_backup(),
|
|
}
|
|
|
|
if any(cast(str, checks[name]["status"]) == "fail" for name in CRITICAL_CHECKS):
|
|
overall_status = "unhealthy"
|
|
elif any(cast(str, check["status"]) == "fail" for check in checks.values()):
|
|
overall_status = "degraded"
|
|
else:
|
|
overall_status = "ok"
|
|
|
|
payload = {
|
|
"status": overall_status,
|
|
"version": {
|
|
"git_sha": os.environ.get("GIT_SHA", "unknown"),
|
|
"build": os.environ.get("BUILD_ID", "unknown"),
|
|
},
|
|
"checks": checks,
|
|
"timestamp": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
}
|
|
response_status = 503 if overall_status == "unhealthy" else 200
|
|
return JsonResponse(payload, status=response_status)
|