From 05aac052f42e9d5b47d60419297ee4d751931803 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:52:21 +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. --- src/endpoint_monitor/logging.py | 36 +++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 src/endpoint_monitor/logging.py diff --git a/src/endpoint_monitor/logging.py b/src/endpoint_monitor/logging.py new file mode 100644 index 0000000..819a8f1 --- /dev/null +++ b/src/endpoint_monitor/logging.py @@ -0,0 +1,36 @@ +import json +import logging +from datetime import UTC, datetime +from urllib.parse import urlsplit, urlunsplit + + +_RESERVED = {"name", "msg", "args", "levelname", "levelno", "pathname", "filename", "module", "exc_info", "exc_text", "stack_info", "lineno", "funcName", "created", "msecs", "relativeCreated", "thread", "threadName", "processName", "process", "taskName"} + + +def redact_url(value: str) -> str: + try: + parsed = urlsplit(value) + host = parsed.hostname or "" + if ":" in host and not host.startswith("["): + host = f"[{host}]" + port = f":{parsed.port}" if parsed.port else "" + return urlunsplit((parsed.scheme, host + port, parsed.path or "/", "", "")) + except (TypeError, ValueError): + return "" + + +class JsonFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + data = {"timestamp": datetime.now(UTC).isoformat(), "level": record.levelname, "event": record.getMessage()} + for key, value in record.__dict__.items(): + if key not in _RESERVED and not key.startswith("_"): + data[key] = value + return json.dumps(data, default=str, separators=(",", ":")) + + +def configure_logging(level: str) -> None: + handler = logging.StreamHandler() + handler.setFormatter(JsonFormatter()) + root = logging.getLogger() + root.handlers[:] = [handler] + root.setLevel(level)