Compare commits

...
3 Commits
Author SHA1 Message Date
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
LeoVasanko c869783b0b Build/dev emoji logging improved. Restored old plain lines, using intentionally different emoji than the main package. 2026-09-18 17:36:17 +00:00
LeoVasanko c7577e9ff7 Guard access log lines against unreset injected colors
App-supplied extra fields may carry raw ANSI color codes without a
reset, bleeding into the next line's IP column and following output.
Ensure access log lines begin and end with a reset, adding one only
where missing to avoid duplicate resets on well-formed lines.
2026-09-18 11:07:36 +00:00
7 changed files with 53 additions and 44 deletions

No files matched your search

+2 -1
View File
@@ -9,7 +9,7 @@ import time
from ipaddress import IPv6Address from ipaddress import IPv6Address
from typing import TYPE_CHECKING, cast from typing import TYPE_CHECKING, cast
from .termwidth import pad_display from .tty import pad_display, use_color
if TYPE_CHECKING: if TYPE_CHECKING:
from uvicorn._types import ( from uvicorn._types import (
@@ -388,6 +388,7 @@ class AccessLogMiddleware:
if message["type"] == "websocket.accept" and not accepted: if message["type"] == "websocket.accept" and not accepted:
accepted = True accepted = True
ws_id = _next_ws_id() 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()) fields = _ws_open_extra(www_scope, ws_id, origin, _extra())
logger.info(_assemble_access_log(fields), extra=fields) logger.info(_assemble_access_log(fields), extra=fields)
elif message["type"] == "websocket.http.response.start" and not closed: elif message["type"] == "websocket.http.response.start" and not closed:
+10 -24
View File
@@ -10,8 +10,6 @@ from __future__ import annotations
import io import io
import logging import logging
import os
import re
import sys import sys
from contextlib import suppress from contextlib import suppress
from copy import deepcopy from copy import deepcopy
@@ -32,34 +30,15 @@ if TYPE_CHECKING:
from .accesslog import AccessLogMiddleware from .accesslog import AccessLogMiddleware
from .environ import env from .environ import env
from .tty import strip_ansi, use_color
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m") RESET = "\033[0m"
ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s" ACCESS_LOG_FMT = "%(client)s %(status)s %(method)s %(host)s%(path)s %(extra)s%(timing)s"
ACCESS_LOGGER = "fastapi_vue.access" 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 = { _LEVEL_EMOJI = {
logging.DEBUG: "🐛", logging.DEBUG: "🐛",
logging.INFO: "🔷", logging.INFO: "🔷",
@@ -122,7 +101,14 @@ class Formatter(logging.Formatter):
return _level_prefix(record) + record.getMessage() return _level_prefix(record) + record.getMessage()
formatted = super().formatMessage(record) formatted = super().formatMessage(record)
if not self.use_colors: if not self.use_colors:
formatted = strip_ansi(formatted) return strip_ansi(formatted)
# Guard against app-supplied fields (``extra``) carrying raw color
# codes without a reset: ensure the line begins and ends with a
# reset, but only add one where it's missing to avoid duplicates.
if not formatted.startswith(RESET):
formatted = RESET + formatted
if not formatted.endswith(RESET):
formatted += RESET
return formatted return formatted
+1 -1
View File
@@ -22,9 +22,9 @@ from .logging import (
patch_lifespan_logging, patch_lifespan_logging,
patch_log_config, patch_log_config,
patch_server_error_middleware, patch_server_error_middleware,
use_color,
) )
from .startupbox import print_box 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 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 import sys
from .termwidth import display_width, pad_display from .tty import display_width, pad_display
def print_box(text: str) -> None: def print_box(text: str) -> None:
"""Print text in a rounded unicode box sized to its longest line. """Print text in a rounded unicode box sized to its longest line.
The text is split on newlines as-is (not trimmed). Padding accounts 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") lines = text.split("\n")
width = max(map(display_width, lines)) 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 re
import sys
import unicodedata import unicodedata
from contextlib import suppress
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]") 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: def _is_wide(char: str) -> bool:
"""Return True for characters rendered as two terminal columns.""" """Return True for characters rendered as two terminal columns."""
if unicodedata.east_asian_width(char) in {"F", "W"}: if unicodedata.east_asian_width(char) in {"F", "W"}:
+10 -14
View File
@@ -10,23 +10,19 @@ from pathlib import Path
MIN_NODE_VERSION = 20 MIN_NODE_VERSION = 20
# Duplicated from fastapi_vue.logging because build environment is isolated
_LEVEL_EMOJI = {
logging.DEBUG: "🐛",
logging.INFO: "🔷",
logging.WARNING: "❗",
logging.ERROR: "🛑",
logging.CRITICAL: "🚨",
}
class _Formatter(logging.Formatter): class _Formatter(logging.Formatter):
"""Emoji level prefix formatter, mirroring fastapi_vue.logging.Formatter.""" """Prefix formatter, intentionally different from fastapi_vue.logging.
INFO and below pass through unprefixed so messages can use their own
markings (>>>, ###); WARNING and above get an emoji prefix.
"""
def format(self, record: logging.LogRecord) -> str: def format(self, record: logging.LogRecord) -> str:
emoji = _LEVEL_EMOJI.get(record.levelno) if record.levelno >= logging.ERROR:
prefix = f"{emoji} " if emoji else f"{record.levelname}: " return f"🛑 {record.getMessage()}"
return prefix + record.getMessage() if record.levelno >= logging.WARNING:
return f"💣 {record.getMessage()}"
return record.getMessage()
_handler = logging.StreamHandler() _handler = logging.StreamHandler()
+1 -1
View File
@@ -133,7 +133,7 @@ async def ready(url: str, path: str = "", max_attempts: int = 50) -> None:
for attempt in range(max_attempts): for attempt in range(max_attempts):
if await http_get_server(f"{url}{path}", timeout=1.0) is not None: if await http_get_server(f"{url}{path}", timeout=1.0) is not None:
logger.info("✓ Backend ready!") logger.info("🟢 Backend ready!")
return return
if attempt == max_attempts - 1: if attempt == max_attempts - 1:
logger.error("Backend at %s didn't start in time", url) logger.error("Backend at %s didn't start in time", url)