Startup box: server.run() prints app name, version and connect URL in a rounded unicode box on stderr. Shared termwidth module for ANSI-aware unicode display width and padding.
This commit is contained in:
@@ -6,10 +6,11 @@ import http
|
||||
import itertools
|
||||
import logging
|
||||
import time
|
||||
import unicodedata
|
||||
from ipaddress import IPv6Address
|
||||
from typing import TYPE_CHECKING, cast
|
||||
|
||||
from .termwidth import pad_display
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from uvicorn._types import (
|
||||
ASGI3Application,
|
||||
@@ -39,14 +40,6 @@ _WS_OPEN = "\033[38;5;226m" # WebSocket connect (brightest yellow)
|
||||
_WS_CLOSE = "\033[38;5;142m" # WebSocket disconnect (dimmer yellow)
|
||||
|
||||
|
||||
def _display_width(text: str) -> int:
|
||||
return sum(2 if unicodedata.east_asian_width(char) in ("F", "W") else 1 for char in text)
|
||||
|
||||
|
||||
def _pad_display(text: str, width: int) -> str:
|
||||
return text + " " * max(width - _display_width(text), 0)
|
||||
|
||||
|
||||
def _format_duration(duration: float) -> str:
|
||||
ms = int(duration * 1000)
|
||||
if ms < 2000:
|
||||
@@ -193,9 +186,9 @@ def _http_access_log_extra(
|
||||
"client": _format_client_ip(client_addr).ljust(19),
|
||||
"status": f"{_status_color(status)}{str(status).rjust(3)}{_RESET}",
|
||||
"method": (
|
||||
f"{_METHOD_READ}{_pad_display('🔌', 7)}{_RESET}"
|
||||
f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}"
|
||||
if method == "🔌"
|
||||
else f"{_method_color(method)}{_pad_display(method, 7)}{_RESET}"
|
||||
else f"{_method_color(method)}{pad_display(method, 7)}{_RESET}"
|
||||
),
|
||||
"host": f"{_HOST}{_header(scope, 'host') or '-'}{_RESET}",
|
||||
"path": f"{_PATH}{full_path}{_RESET}",
|
||||
@@ -229,7 +222,7 @@ def _ws_open_extra(
|
||||
return {
|
||||
"client": _format_client_ip(client_addr).ljust(19),
|
||||
"status": f"{_WS_OPEN} {ws_id}{_RESET}",
|
||||
"method": f"{_METHOD_READ}{_pad_display('🔌', 7)}{_RESET}",
|
||||
"method": f"{_METHOD_READ}{pad_display('🔌', 7)}{_RESET}",
|
||||
"host": f"{_HOST}{host}{_RESET}" if host else "",
|
||||
"path": path,
|
||||
"extra": extra,
|
||||
@@ -264,7 +257,7 @@ def _ws_close_extra(
|
||||
return {
|
||||
"client": " " * 19,
|
||||
"status": f"{_WS_CLOSE} {ws_id}{_RESET}",
|
||||
"method": f"{_TIMING}{_pad_display('closed', 7)}{_RESET}",
|
||||
"method": f"{_TIMING}{pad_display('closed', 7)}{_RESET}",
|
||||
"host": "",
|
||||
"path": f"{code} {status_text}",
|
||||
"extra": extra,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Uvicorn server runner with multi-endpoint support."""
|
||||
|
||||
import asyncio
|
||||
import importlib.metadata
|
||||
import logging
|
||||
import os
|
||||
from contextlib import suppress
|
||||
@@ -13,10 +14,55 @@ from uvicorn import Config, Server
|
||||
|
||||
from .hostutil import parse_endpoints
|
||||
from .logging import install_access_log, patch_log_config
|
||||
from .startupbox import print_box
|
||||
|
||||
tracerite.load() # Early load on CLI load (import server); uvicorn workers reload via log config
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_WILDCARD_HOSTS = frozenset({"0.0.0.0", "::"}) # noqa: S104
|
||||
|
||||
|
||||
def _connect_url(endpoints: list[dict]) -> str:
|
||||
"""Return a URL the user can connect to for the first TCP endpoint.
|
||||
|
||||
Wildcard binds (0.0.0.0, ::) are shown as localhost, as that is the
|
||||
address a user can actually open. Returns "" for unix-socket-only setups.
|
||||
"""
|
||||
for endpoint in endpoints:
|
||||
host = endpoint.get("host")
|
||||
if host is None:
|
||||
continue
|
||||
if host in _WILDCARD_HOSTS:
|
||||
host = "localhost"
|
||||
elif ":" in host: # IPv6 literal
|
||||
host = f"[{host}]"
|
||||
return f"http://{host}:{endpoint['port']}"
|
||||
return ""
|
||||
|
||||
|
||||
def _print_startup_box(template: str, app: str, endpoints: list[dict]) -> None:
|
||||
"""Format the startup box template and print it.
|
||||
|
||||
Available fields: ``{module}`` (top-level package of the app path),
|
||||
``{name}`` (module with spaces instead of underscores), ``{Name}``
|
||||
(also capitalized), ``{version}`` (from installed package metadata,
|
||||
"dev" when not installed) and ``{url}``.
|
||||
"""
|
||||
module = app.split(":", 1)[0].split(".", 1)[0]
|
||||
name = module.replace("_", " ")
|
||||
try:
|
||||
version = importlib.metadata.version(module)
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
version = ""
|
||||
values = {
|
||||
"module": module,
|
||||
"name": name,
|
||||
"Name": name.title(),
|
||||
"version": version,
|
||||
"url": _connect_url(endpoints),
|
||||
}
|
||||
print_box(template.format_map(values))
|
||||
|
||||
|
||||
def run( # noqa: PLR0913
|
||||
app: str,
|
||||
@@ -26,6 +72,7 @@ def run( # noqa: PLR0913
|
||||
reload: bool | Path = False,
|
||||
workers: int | None = None,
|
||||
access_log: bool = True,
|
||||
startup_box: str | None = "{Name} {version}\n{url}",
|
||||
log_config: Any = uvicorn.config.LOGGING_CONFIG, # noqa: ANN401
|
||||
**uvicorn_config: Any, # noqa: ANN401
|
||||
) -> None:
|
||||
@@ -41,6 +88,8 @@ def run( # noqa: PLR0913
|
||||
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
|
||||
access_log: Enable our colored HTTP/WebSocket access logging middleware
|
||||
(uvicorn's own access logging is always bypassed).
|
||||
startup_box: Template for the startup box printed to stderr before
|
||||
serving (see _print_startup_box for fields), None to not print it.
|
||||
log_config: Logging config passed to uvicorn. Dict configs are patched
|
||||
best-effort (see fastapi_vue.logging.patch_log_config): tracerite
|
||||
loading and WebSocket chatter filtering are always installed, and
|
||||
@@ -53,6 +102,9 @@ def run( # noqa: PLR0913
|
||||
msg = "No endpoints to serve; check listen configuration"
|
||||
raise ValueError(msg)
|
||||
|
||||
if startup_box:
|
||||
_print_startup_box(startup_box, app, endpoints)
|
||||
|
||||
if isinstance(reload, Path):
|
||||
uvicorn_config["reload_dirs"] = [str(reload)]
|
||||
elif not reload:
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
"""Startup banner: a rounded unicode box printed to stderr (not via logging)."""
|
||||
|
||||
import sys
|
||||
|
||||
from .termwidth 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).
|
||||
"""
|
||||
lines = text.split("\n")
|
||||
width = max(map(display_width, lines))
|
||||
border = "─" * (width + 2)
|
||||
out = [f"╭{border}╮"]
|
||||
out.extend(f"│ {pad_display(line, width)} │" for line in lines)
|
||||
out.append(f"╰{border}╯")
|
||||
sys.stderr.write("\n".join(out) + "\n")
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Terminal display width calculation for unicode and ANSI-colored text."""
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;:]*[A-Za-z]")
|
||||
|
||||
|
||||
def _is_wide(char: str) -> bool:
|
||||
"""Return True for characters rendered as two terminal columns."""
|
||||
if unicodedata.east_asian_width(char) in {"F", "W"}:
|
||||
return True
|
||||
cp = ord(char)
|
||||
return unicodedata.category(char) == "So" and (
|
||||
0x2600 <= cp <= 0x27BF or 0x1F300 <= cp <= 0x1F9FF or 0x1FA00 <= cp <= 0x1FAFF
|
||||
)
|
||||
|
||||
|
||||
def display_width(text: str) -> int:
|
||||
"""Calculate the display width of a string in terminal columns.
|
||||
|
||||
ANSI escape codes are ignored. Wide characters (East Asian F/W and
|
||||
emoji) count as two columns, combining marks and format characters
|
||||
(e.g. emoji variation selectors) as zero.
|
||||
"""
|
||||
plain = ANSI_ESCAPE_RE.sub("", text)
|
||||
width = 0
|
||||
for char in plain:
|
||||
if unicodedata.category(char) in {"Mn", "Mc", "Me", "Cf"}:
|
||||
continue
|
||||
width += 2 if _is_wide(char) else 1
|
||||
return width
|
||||
|
||||
|
||||
def pad_display(text: str, width: int) -> str:
|
||||
"""Pad text with trailing spaces to the given display width (no truncation)."""
|
||||
return text + " " * max(width - display_width(text), 0)
|
||||
Reference in New Issue
Block a user