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.
This commit is contained in:
1 parent
c869783b0b
commit
465206067b
5 files changed
+33
-29
No files matched your search
@@ -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:
|
||||
|
||||
@@ -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: "🔷",
|
||||
|
||||
@@ -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,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"}:
|
||||
Reference in new issue
Block a user