Files
OrbitWard/worker/app/collectors/website.py
T
2026-05-22 17:36:40 -06:00

51 lines
1.7 KiB
Python

from dataclasses import dataclass
from time import perf_counter
import httpx
@dataclass(frozen=True)
class WebsiteCheckConfig:
url: str
expected_status: int = 200
expected_text: str | None = None
unexpected_text: str | None = None
timeout_seconds: float = 10.0
@dataclass(frozen=True)
class WebsiteCheckResult:
status: str
response_time_ms: int | None
message: str
async def run_website_check(config: WebsiteCheckConfig) -> WebsiteCheckResult:
started = perf_counter()
try:
async with httpx.AsyncClient(follow_redirects=True, timeout=config.timeout_seconds) as client:
response = await client.get(config.url)
except httpx.HTTPError as exc:
return WebsiteCheckResult(status="down", response_time_ms=None, message=str(exc))
response_time_ms = int((perf_counter() - started) * 1000)
if response.status_code != config.expected_status:
return WebsiteCheckResult(
status="down",
response_time_ms=response_time_ms,
message=f"Expected HTTP {config.expected_status}, got {response.status_code}",
)
if config.expected_text and config.expected_text not in response.text:
return WebsiteCheckResult(
status="down",
response_time_ms=response_time_ms,
message="Expected text was not present",
)
if config.unexpected_text and config.unexpected_text in response.text:
return WebsiteCheckResult(
status="down",
response_time_ms=response_time_ms,
message="Unexpected text was present",
)
return WebsiteCheckResult(status="up", response_time_ms=response_time_ms, message="Website check passed")