From 16da6032bfb367b5ef5bf09fdd262014e3e332a5 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:57:44 +0000 Subject: [PATCH] decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project. --- app/checker.py | 108 ++++++++++++++++++++++++++++++------------------- 1 file changed, 67 insertions(+), 41 deletions(-) diff --git a/app/checker.py b/app/checker.py index 345dc6f..0632d1d 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,50 +1,76 @@ import logging -from time import monotonic +import time +from datetime import UTC, datetime from urllib.parse import urljoin import httpx -from .config import Settings -from .logging import log_event, redacted_url -from .models import CheckSnapshot, State, now -from .security import UnsafeDestination, validate_destination +from app.config import Settings +from app.models import MonitorState, StatusSnapshot +from app.security import Resolver, UnsafeTarget, redact_url, system_resolver, validate_target -logger = logging.getLogger("endpoint_monitor.checker") +logger = logging.getLogger("monitor.checker") -async def check_url(url: str, timeout: float, settings: Settings, - transport: httpx.AsyncBaseTransport | None = None) -> CheckSnapshot: - started = monotonic() - current = url - try: - async with httpx.AsyncClient(transport=transport, follow_redirects=False, - timeout=httpx.Timeout(timeout)) as client: - for redirect_count in range(settings.max_redirects + 1): - await validate_destination(current) - response = await client.get(current) - if response.is_redirect: - location = response.headers.get("location") - if not location: - break - if redirect_count == settings.max_redirects: - raise httpx.TooManyRedirects("redirect limit exceeded") - current = urljoin(current, location) - continue - elapsed = round((monotonic() - started) * 1000, 3) - state = State.up if 200 <= response.status_code < 400 else State.down - result = CheckSnapshot(state=state, checked_at=now(), latency_ms=elapsed, - http_status=response.status_code) - log_event(logger, "check_complete", url=redacted_url(current), state=state, - latency_ms=elapsed, http_status=response.status_code) - return result - raise httpx.RemoteProtocolError("redirect response missing location") - except UnsafeDestination: - log_event(logger, "check_blocked", url=redacted_url(current)) - raise - except (httpx.HTTPError, TimeoutError) as exc: - elapsed = round((monotonic() - started) * 1000, 3) - result = CheckSnapshot(state=State.error, checked_at=now(), latency_ms=elapsed, - error=type(exc).__name__) - log_event(logger, "check_error", url=redacted_url(current), state=State.error, - error=type(exc).__name__, latency_ms=elapsed) +class EndpointChecker: + def __init__( + self, settings: Settings, resolver: Resolver = system_resolver, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self.settings = settings + self.resolver = resolver + self.transport = transport + + async def check(self, monitor_id: str, url: str) -> StatusSnapshot: + started = time.perf_counter() + current = url + try: + timeout = httpx.Timeout( + self.settings.check_timeout_seconds, + connect=self.settings.connect_timeout_seconds, + ) + async with httpx.AsyncClient( + timeout=timeout, transport=self.transport, follow_redirects=False, + ) as client: + for hop in range(self.settings.max_redirects + 1): + await validate_target(current, self.resolver) + async with client.stream("GET", current) as response: + if response.is_redirect: + location = response.headers.get("location") + if not location: + return self._result(MonitorState.ERROR, started, error="bad_redirect") + if hop == self.settings.max_redirects: + return self._result(MonitorState.ERROR, started, error="too_many_redirects") + current = urljoin(current, location) + continue + state = MonitorState.UP if 200 <= response.status_code < 400 else MonitorState.DOWN + result = self._result(state, started, response.status_code) + self._log(monitor_id, url, result) + return result + except UnsafeTarget as exc: + result = self._result(MonitorState.BLOCKED, started, error=str(exc)) + except httpx.TimeoutException: + result = self._result(MonitorState.ERROR, started, error="timeout") + except httpx.HTTPError: + result = self._result(MonitorState.ERROR, started, error="transport_error") + self._log(monitor_id, url, result) return result + + @staticmethod + def _result( + state: MonitorState, started: float, status: int | None = None, + error: str | None = None, + ) -> StatusSnapshot: + return StatusSnapshot( + state=state, checked_at=datetime.now(UTC), + latency_ms=round((time.perf_counter() - started) * 1000, 3), + http_status=status, error=error, + ) + + @staticmethod + def _log(monitor_id: str, url: str, result: StatusSnapshot) -> None: + logger.info("check_complete", extra={ + "monitor_id": monitor_id, "target": redact_url(url), + "state": result.state, "latency_ms": result.latency_ms, + "http_status": result.http_status, "error": result.error, + })