Refine health checks and add polling animation

This commit is contained in:
2025-11-27 11:43:35 +01:00
parent c841e27c30
commit 25d80f5723
18 changed files with 499 additions and 219 deletions

View File

@@ -0,0 +1,89 @@
import asyncio
from datetime import datetime, timezone
from core.config import Settings
from operations.health.schemas import ComponentCheck, HealthStatus
async def check_database_status() -> ComponentCheck:
"""Checking the primary database connection with a timeout."""
try:
# Simulate an async DB call (replace with actual logic)
await asyncio.wait_for(asyncio.sleep(0.042), timeout=0.1)
return ComponentCheck(name = "postgres",
status = "pass",
time = datetime.now(timezone.utc),
observedValue = 42,
observedUnit = "ms")
except asyncio.TimeoutError:
return ComponentCheck(name = "postgres",
status = "fail",
time = datetime.now(timezone.utc),
output = "Check timed out")
except Exception as e:
return ComponentCheck(name = "postgres",
status = "fail",
time = datetime.now(timezone.utc),
output = str(e))
async def check_media_server_status() -> ComponentCheck:
"""Checking the media server connection with a timeout."""
try:
# Simulate a successful async queue call
await asyncio.wait_for(asyncio.sleep(0.02), timeout=0.05)
return ComponentCheck(name = "livekit",
status = "pass",
time = datetime.now(timezone.utc),
observedValue = 20,
observedUnit = "ms")
except asyncio.TimeoutError:
return ComponentCheck(name = "livekit",
status = "fail",
time = datetime.now(timezone.utc),
output = "Check timed out")
except Exception as e:
return ComponentCheck(name = "livekit",
status = "fail",
time = datetime.now(timezone.utc),
output = str(e))
async def readiness_check() -> bool:
"""
*Extend readiness_check() as dependencies are added; it must return True/False.*
"""
# Run all critical checks
db = await check_database_status()
media_server = await check_media_server_status()
# Check if all statuses are 'pass'
if db.status == 'pass' and media_server.status == 'pass':
return True
else:
return False
async def get_detailed_health(settings: Settings) -> HealthStatus:
"""
Build detailed health payload aligned with common status formats.
status: pass | warn | fail
"""
# Run all critical checks
db = await check_database_status()
media_server = await check_media_server_status()
# Check if all statuses are 'pass'
overall = "pass"
if db.status != 'pass' or media_server.status != 'pass':
overall = "fail"
return HealthStatus(status = overall,
version = settings.version,
environment = settings.environment,
serviceName = settings.service_name,
description = settings.title,
checks = {"Database": db,
"Media Server": media_server})