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"}: