32 lines
955 B
Python
32 lines
955 B
Python
|
"""HTTP health checker for OpsBridge."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class HealthResult:
|
||
|
|
ok: bool
|
||
|
|
status_code: Optional[int] = None
|
||
|
|
error: Optional[str] = None
|
||
|
|
|
||
|
|
|
||
|
|
class HealthChecker:
|
||
|
|
def __init__(self, url: str, timeout_seconds: int = 5):
|
||
|
|
self._url = url
|
||
|
|
self._timeout = timeout_seconds
|
||
|
|
|
||
|
|
async def check(self) -> HealthResult:
|
||
|
|
try:
|
||
|
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
||
|
|
response = await client.get(self._url)
|
||
|
|
response.raise_for_status()
|
||
|
|
return HealthResult(ok=True, status_code=response.status_code)
|
||
|
|
except httpx.HTTPStatusError as e:
|
||
|
|
return HealthResult(ok=False, status_code=e.response.status_code, error=str(e))
|
||
|
|
except Exception as e:
|
||
|
|
return HealthResult(ok=False, error=str(e))
|