Logging cleanup, better color compatibility for Mac Terminal and consistent across DB and FastAPI access logs.

paskia/db/logging.py

@@ -35,9 +35,9 @@ _UNSAFE_CHARS = re.compile(
 
 # ANSI color codes (matching FastAPI logging style)
 _RESET = "\033[0m"
-_DIM = "\033[2m"
+_SEP = "\033[38;5;242m"  # Dark grey for separators (like host/timing in access log)
-_PATH_PREFIX = "\033[1;30m"  # Dark grey for path prefix (like host in access log)
+_PATH_PREFIX = "\033[38;5;242m"  # Dark grey for path prefix (like host in access log)
-_PATH_FINAL = "\033[0m"  # Default for final element (like path in access log)
+_PATH_FINAL = "\033[38;5;250m"  # Default for final element (like path in access log)
 _DELETE = "\033[1;31m"  # Red for deletions
 _ADD = "\033[0;32m"  # Green for additions
 _ACTION = "\033[1;34m"  # Bold blue for action name
@@ -317,7 +317,7 @@ def _format_change_lines(
     # Helper to format a value, checking for censored paths
     def fmt_value(v: Any, child_path: list[str]) -> str:
         if child_path[-2:] == ["oidc", "key"]:
-            return f"{_DIM}<hidden>{_RESET}"
+            return f"{_SEP}<hidden>{_RESET}"
         return _format_value(v, resolver=resolver)
 
     # Helper to format path with UUID replacement
@@ -342,12 +342,12 @@ def _format_change_lines(
             lines = []
             # First line: path with green final element and grey =
             if len(formatted_path) == 1:
-                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_DIM}={_RESET}")
+                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET}")
             else:
                 prefix = ".".join(formatted_path[:-1])
                 final = formatted_path[-1]
                 lines.append(
-                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_DIM}={_RESET}"
+                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}"
                 )
             # Child lines: indented key: value, with aligned values
             # Format keys (may contain UUIDs)
@@ -360,24 +360,24 @@ def _format_change_lines(
             field_width = max(max_key_len, 12)  # minimum 12 chars
             for k_display, v_str in formatted_items:
                 padding = " " * (field_width - len(k_display))
-                lines.append(f"    {k_display}{_DIM}:{_RESET}{padding} {v_str}")
+                lines.append(f"    {k_display}{_SEP}:{_RESET}{padding} {v_str}")
             return lines
         else:
             value_str = fmt_value(value, path)
             if len(formatted_path) == 1:
                 return [
-                    f"  {_ADD}{formatted_path[0]}{_RESET} {_DIM}={_RESET} {value_str}"
+                    f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET} {value_str}"
                 ]
             prefix = ".".join(formatted_path[:-1])
             final = formatted_path[-1]
             return [
-                f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_DIM}={_RESET} {value_str}"
+                f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET} {value_str}"
             ]
 
     # update: Existing item being updated - normal path colors
     value_str = fmt_value(value, path)
     path_str = _format_path(path, resolver=resolver)
-    return [f"  {path_str} {_DIM}={_RESET} {value_str}"]
+    return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
 
 
 def format_diff(

paskia/fastapi/logging.py

@@ -115,25 +115,16 @@ def format_access_log(
     client: str, status: int, method: str, host: str, path: str, duration_ms: float
 ) -> str:
     """Format access log line with colors and aligned fields."""
-    use_color = sys.stderr.isatty()
-
     # Format components with fixed widths for alignment
     ip = format_client_ip(client).ljust(19)  # IPv6 network max 19 chars
     timing = f"{duration_ms:.0f}ms"
     method_padded = method.ljust(7)  # Longest method is OPTIONS (7)
 
-    if use_color:
+    status_str = f"{status_color(status)}{status}{_RESET}"
-        status_str = f"{status_color(status)}{status}{_RESET}"
+    timing_str = f"{_TIMING}{timing}{_RESET}"
-        timing_str = f"{_TIMING}{timing}{_RESET}"
+    method_str = f"{method_color(method)}{method_padded}{_RESET}"
-        method_str = f"{method_color(method)}{method_padded}{_RESET}"
+    host_str = f"{_HOST}{host}{_RESET}"
-        host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
-        path_str = f"{_PATH}{path}{_RESET}"
-    else:
-        status_str = str(status)
-        timing_str = timing
-        method_str = method_padded
-        host_str = host
-        path_str = path
 
     # Format: "IP STATUS METHOD host path TIMING"
     return f"{ip} {status_str} {method_str} {host_str}{path_str} {timing_str}"
@@ -153,7 +144,6 @@ def _next_ws_id() -> int:
 
 def log_ws_open(ws) -> int:
     """Log WebSocket connection open. Returns connection ID for use in close."""
-    use_color = sys.stderr.isatty()
     ws_id = _next_ws_id()
 
     client = ws.client.host if ws.client else "-"
@@ -169,19 +159,11 @@ def log_ws_open(ws) -> int:
     origin_host = origin.split("://", 1)[-1] if origin else None
     show_origin = origin_host and origin_host != host
 
-    if use_color:
+    # 🔌 aligned with status (takes ~2 char width), ID aligned with method
-        # 🔌 aligned with status (takes ~2 char width), ID aligned with method
+    prefix = f"🔌  {_WS_OPEN}{id_str}{_RESET}"
-        prefix = f"🔌  {_WS_OPEN}{id_str}{_RESET}"
+    host_str = f"{_HOST}{host}{_RESET}"
-        host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
-        path_str = f"{_PATH}{path}{_RESET}"
+    origin_str = f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else ""
-        origin_str = (
-            f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else ""
-        )
-    else:
-        prefix = f"WS+ {id_str}"
-        host_str = host
-        path_str = path
-        origin_str = f" from {origin_host}" if show_origin else ""
 
     logger.info(f"{ip} {prefix} {host_str}{path_str}{origin_str}")
     return ws_id
@@ -209,8 +191,6 @@ WS_CLOSE_CODES = {
 
 def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None:
     """Log WebSocket connection close with duration and status."""
-    use_color = sys.stderr.isatty()
-
     id_str = f"{ws_id:02d}".ljust(7)  # Align with method field (7 chars)
     timing = f"{duration * 1000:.0f}ms"
 
@@ -220,15 +200,10 @@ def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None:
     else:
         status = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
 
-    if use_color:
+    # 🔌 aligned with status, ID aligned with method
-        # 🔌 aligned with status, ID aligned with method
+    prefix = f"🔌  {_WS_CLOSE}{id_str}{_RESET}"
-        prefix = f"🔌  {_WS_CLOSE}{id_str}{_RESET}"
+    status_str = f"{_WS_STATUS}{status}{_RESET}"
-        status_str = f"{_WS_STATUS}{status}{_RESET}"
+    timing_str = f"{_TIMING}{timing}{_RESET}"
-        timing_str = f"{_TIMING}{timing}{_RESET}"
-    else:
-        prefix = f"WS- {id_str}"
-        status_str = status
-        timing_str = timing
 
     logger.info(f"{' ' * 19} {prefix} {status_str} {timing_str}")