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

@@ -1,89 +1,107 @@
 import logging
-from time import perf_counter
+import time
+from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
 from urllib.parse import urljoin
 
 import httpx
 
-from app.models import CheckResult, CurrentStatus, Monitor, State, now_utc
+from app.config import Settings
-from app.security import Resolver, UnsafeTargetError, redacted_url, resolve_addresses, validate_target
+from app.models import CurrentStatus, State
+from app.security import (
+    DestinationRejected,
+    Resolver,
+    redacted_url,
+    require_public_destination,
+    system_resolver,
+)
 
 logger = logging.getLogger(__name__)
-REDIRECTS = {301, 302, 303, 307, 308}
+RedirectCodes = {301, 302, 303, 307, 308}
 
 
 class EndpointChecker:
     def __init__(
         self,
-        timeout_seconds: float,
+        settings: Settings,
-        max_redirects: int,
+        resolver: Resolver = system_resolver,
-        resolver: Resolver = resolve_addresses,
         transport: httpx.AsyncBaseTransport | None = None,
     ) -> None:
-        self.timeout_seconds = timeout_seconds
+        self.settings = settings
-        self.max_redirects = max_redirects
         self.resolver = resolver
         self.transport = transport
 
-    async def check(self, monitor: Monitor) -> CheckResult:
+    async def check(self, monitor_id: str, url: str) -> CurrentStatus:
-        started = perf_counter()
+        started = time.monotonic()
-        url = str(monitor.url)
+        safe_url = redacted_url(url)
-        state = State.ERROR
-        observed: int | None = None
-        error: str | None = None
         try:
-            async with httpx.AsyncClient(
+            result = await self._request(url, started)
-                transport=self.transport,
+        except DestinationRejected:
-                timeout=httpx.Timeout(self.timeout_seconds),
+            self._log(monitor_id, safe_url, State.ERROR, started, None)
-                follow_redirects=False,
+            raise
-            ) as client:
+        except (httpx.TimeoutException, httpx.NetworkError, httpx.ProtocolError) as exc:
-                for hop in range(self.max_redirects + 1):
+            result = self._error(started, self._safe_error(exc))
-                    await validate_target(url, self.resolver)
+        self._log(monitor_id, safe_url, result.state, started, result.status_code)
-                    async with client.stream(
+        return result
-                        "GET", url, headers={"user-agent": "endpoint-monitor/0.1"}
-                    ) as response:
-                        observed = response.status_code
-                        if observed not in REDIRECTS:
-                            state = (
-                                State.UP if observed == monitor.expected_status else State.DOWN
-                            )
-                            break
-                        location = response.headers.get("location")
-                        if not location:
-                            error = "redirect response omitted Location"
-                            break
-                        if hop == self.max_redirects:
-                            error = "redirect limit exceeded"
-                            break
-                        url = urljoin(url, location)
-        except UnsafeTargetError:
-            state = State.BLOCKED
-            error = "target blocked by outbound request policy"
-        except httpx.TimeoutException:
-            error = "outbound request timed out"
-        except httpx.HTTPError:
-            error = "outbound HTTP request failed"
-        except OSError:
-            error = "outbound network operation failed"
 
-        latency = round((perf_counter() - started) * 1000, 3)
+    async def _request(self, initial_url: str, started: float) -> CurrentStatus:
-        status = CurrentStatus(
+        timeout = httpx.Timeout(
-            state=state,
+            self.settings.request_timeout_seconds,
-            checked_at=now_utc(),
+            connect=self.settings.connect_timeout_seconds,
-            observed_status=observed,
-            latency_ms=latency,
-            error=error,
         )
+        current = initial_url
+        async with httpx.AsyncClient(
+            timeout=timeout, follow_redirects=False, transport=self.transport
+        ) as client:
+            for hop in range(self.settings.max_redirects + 1):
+                await require_public_destination(current, self.resolver)
+                async with client.stream("GET", current) as response:
+                    if response.status_code not in RedirectCodes:
+                        state = State.UP if 200 <= response.status_code < 400 else State.DOWN
+                        return CurrentStatus(
+                            state=state,
+                            checked_at=datetime.now(UTC),
+                            status_code=response.status_code,
+                            latency_ms=self._elapsed(started),
+                        )
+                    location = response.headers.get("location")
+                if not location:
+                    return self._error(started, "redirect response omitted Location")
+                if hop == self.settings.max_redirects:
+                    return self._error(started, "redirect limit exceeded")
+                current = urljoin(current, location)
+        return self._error(started, "redirect limit exceeded")  # pragma: no cover
+
+    @staticmethod
+    def _elapsed(started: float) -> float:
+        return round((time.monotonic() - started) * 1000, 3)
+
+    def _error(self, started: float, message: str) -> CurrentStatus:
+        return CurrentStatus(
+            state=State.ERROR,
+            checked_at=datetime.now(UTC),
+            latency_ms=self._elapsed(started),
+            error=message[:200],
+        )
+
+    @staticmethod
+    def _safe_error(exc: Exception) -> str:
+        if isinstance(exc, httpx.TimeoutException):
+            return "outbound request timed out"
+        return f"outbound request failed: {type(exc).__name__}"
+
+    @staticmethod
+    def _log(
+        monitor_id: str, url: str, state: State, started: float, status_code: int | None
+    ) -> None:
         logger.info(
-            "endpoint_check",
+            "endpoint check completed",
             extra={
-                "event_data": {
+                "event": "endpoint_check",
-                    "event": "endpoint_check",
+                "monitor_id": monitor_id,
-                    "monitor_id": str(monitor.id),
+                "url": url,
-                    "url": redacted_url(str(monitor.url)),
+                "state": state,
-                    "state": state.value,
+                "latency_ms": EndpointChecker._elapsed(started),
-                    "observed_status": observed,
+                "status_code": status_code,
-                    "latency_ms": latency,
-                }
             },
         )
-        return CheckResult(monitor_id=monitor.id, **status.model_dump())