Compare commits

...
2 Commits
Author SHA1 Message Date
LeoVasanko a19c9028c5 Detect color support on stderr, matching where logs are written 2026-10-21 19:49:20 +00:00
LeoVasanko 465206067b 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.
2026-10-21 19:40:24 +00:00
5 changed files with 34 additions and 31 deletions

No files matched your search

+2 -1
View File
@@ -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:
+2 -26
View File
@@ -10,9 +10,6 @@ from __future__ import annotations
import io
import logging
import os
import re
import sys
from contextlib import suppress
from copy import deepcopy
from typing import TYPE_CHECKING, Literal
@@ -32,8 +29,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 +38,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: "🔷",
@@ -115,7 +91,7 @@ class Formatter(logging.Formatter):
if use_colors in (True, False):
self.use_colors = use_colors
else:
self.use_colors = use_color(sys.stdout)
self.use_colors = use_color()
super().__init__(fmt=fmt, datefmt=datefmt, style=style)
def formatMessage(self, record: logging.LogRecord) -> str: # noqa: N802
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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))
@@ -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"}: