From 465206067b425919b1c7330077db55b8bc0e31b5 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Wed, 21 Oct 2026 19:40:24 +0000 Subject: [PATCH] Expose WebSocket connection id as ws.state.logid on accept Store the two-digit WS id (colored when stderr supports color, plain otherwise) in the ASGI scope state so user code can include it in its own log lines via websocket.state.logid. Move terminal helpers (use_color, strip_ansi, display width) into a new dependency-free fastapi_vue.tty module, absorbing termwidth.py, to keep imports one-directional and avoid the logging/accesslog cycle. --- fastapi-vue/fastapi_vue/accesslog.py | 3 +- fastapi-vue/fastapi_vue/logging.py | 25 +---------------- fastapi-vue/fastapi_vue/server.py | 2 +- fastapi-vue/fastapi_vue/startupbox.py | 4 +-- .../fastapi_vue/{termwidth.py => tty.py} | 28 ++++++++++++++++++- 5 files changed, 33 insertions(+), 29 deletions(-) rename fastapi-vue/fastapi_vue/{termwidth.py => tty.py} (55%) diff --git a/fastapi-vue/fastapi_vue/accesslog.py b/fastapi-vue/fastapi_vue/accesslog.py index 4954389..1f27a63 100644 --- a/fastapi-vue/fastapi_vue/accesslog.py +++ b/fastapi-vue/fastapi_vue/accesslog.py @@ -9,7 +9,7 @@ import time from ipaddress import IPv6Address from typing import TYPE_CHECKING, cast -from .termwidth import pad_display +from .tty import pad_display, use_color if TYPE_CHECKING: from uvicorn._types import ( @@ -388,6 +388,7 @@ class AccessLogMiddleware: if message["type"] == "websocket.accept" and not accepted: accepted = True ws_id = _next_ws_id() + www_scope["state"]["logid"] = f"{_WS_OPEN}{ws_id}{_RESET}" if use_color() else ws_id fields = _ws_open_extra(www_scope, ws_id, origin, _extra()) logger.info(_assemble_access_log(fields), extra=fields) elif message["type"] == "websocket.http.response.start" and not closed: diff --git a/fastapi-vue/fastapi_vue/logging.py b/fastapi-vue/fastapi_vue/logging.py index 38a5a0f..c00acd3 100644 --- a/fastapi-vue/fastapi_vue/logging.py +++ b/fastapi-vue/fastapi_vue/logging.py @@ -10,8 +10,6 @@ from __future__ import annotations import io import logging -import os -import re import sys from contextlib import suppress from copy import deepcopy @@ -32,8 +30,7 @@ if TYPE_CHECKING: from .accesslog import AccessLogMiddleware from .environ import env - -ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m") +from .tty import strip_ansi, use_color RESET = "\033[0m" @@ -42,26 +39,6 @@ ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(t ACCESS_LOGGER = "fastapi_vue.access" -def strip_ansi(text: str) -> str: - """Remove ANSI escape codes from text.""" - return ANSI_ESCAPE_RE.sub("", text) - - -def use_color(stream: io.TextIOBase = sys.stderr) -> bool: - """Test if the stream supports color codes.""" - if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org) - return False - if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node - return True - if hasattr(stream, "isatty") and stream.isatty(): - return True - with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat) - dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1)) - st = os.fstat(stream.fileno()) - return st.st_dev == dev and st.st_ino == ino - return False - - _LEVEL_EMOJI = { logging.DEBUG: "🐛", logging.INFO: "🔷", diff --git a/fastapi-vue/fastapi_vue/server.py b/fastapi-vue/fastapi_vue/server.py index cf62633..8fa5543 100644 --- a/fastapi-vue/fastapi_vue/server.py +++ b/fastapi-vue/fastapi_vue/server.py @@ -22,9 +22,9 @@ from .logging import ( patch_lifespan_logging, patch_log_config, patch_server_error_middleware, - use_color, ) from .startupbox import print_box +from .tty import use_color tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config diff --git a/fastapi-vue/fastapi_vue/startupbox.py b/fastapi-vue/fastapi_vue/startupbox.py index db5c8e1..93de9fa 100644 --- a/fastapi-vue/fastapi_vue/startupbox.py +++ b/fastapi-vue/fastapi_vue/startupbox.py @@ -2,14 +2,14 @@ import sys -from .termwidth import display_width, pad_display +from .tty import display_width, pad_display def print_box(text: str) -> None: """Print text in a rounded unicode box sized to its longest line. The text is split on newlines as-is (not trimmed). Padding accounts - for ANSI colors and unicode display width (see termwidth.display_width). + for ANSI colors and unicode display width (see tty.display_width). """ lines = text.split("\n") width = max(map(display_width, lines)) diff --git a/fastapi-vue/fastapi_vue/termwidth.py b/fastapi-vue/fastapi_vue/tty.py similarity index 55% rename from fastapi-vue/fastapi_vue/termwidth.py rename to fastapi-vue/fastapi_vue/tty.py index 8a4b4f4..0e1da42 100644 --- a/fastapi-vue/fastapi_vue/termwidth.py +++ b/fastapi-vue/fastapi_vue/tty.py @@ -1,11 +1,37 @@ -"""Terminal display width calculation for unicode and ANSI-colored text.""" +"""Terminal color detection, ANSI helpers, and display width calculation.""" +from __future__ import annotations + +import io +import os import re +import sys import unicodedata +from contextlib import suppress ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]") +def strip_ansi(text: str) -> str: + """Remove ANSI escape codes from text.""" + return ANSI_ESCAPE_RE.sub("", text) + + +def use_color(stream: io.TextIOBase = sys.stderr) -> bool: + """Test if the stream supports color codes.""" + if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org) + return False + if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node + return True + if hasattr(stream, "isatty") and stream.isatty(): + return True + with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat) + dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1)) + st = os.fstat(stream.fileno()) + return st.st_dev == dev and st.st_ino == ino + return False + + def _is_wide(char: str) -> bool: """Return True for characters rendered as two terminal columns.""" if unicodedata.east_asian_width(char) in {"F", "W"}: