Compare commits
2
Commits
v1.7.2
..
a19c9028c5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a19c9028c5 | ||
|
|
465206067b |
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,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
|
||||
|
||||
@@ -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