Updates
This commit is contained in:
@@ -0,0 +1,372 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from typing import Iterable
|
||||
|
||||
try:
|
||||
from rich.console import Console
|
||||
from rich.logging import RichHandler
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
except ImportError: # pragma: no cover - exercised only before dependency install
|
||||
Console = None
|
||||
RichHandler = None
|
||||
Table = None
|
||||
Text = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LoggingSettings:
|
||||
app_name: str
|
||||
app_env: str
|
||||
host: str
|
||||
port: int
|
||||
log_level: str
|
||||
log_verbose: bool
|
||||
database_url: str
|
||||
version: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StartupStatus:
|
||||
app_name: str
|
||||
version: str
|
||||
environment: str
|
||||
host: str
|
||||
port: int
|
||||
database: str
|
||||
mode: str
|
||||
started_at: str
|
||||
local_url: str
|
||||
network_url: str
|
||||
|
||||
|
||||
class PlainFormatter(logging.Formatter):
|
||||
default_time_format = "%Y-%m-%d %H:%M:%S"
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
if not hasattr(record, "component"):
|
||||
record.component = record.name.rsplit(".", 1)[-1]
|
||||
return super().format(record)
|
||||
|
||||
|
||||
def _allow_color() -> bool:
|
||||
if RichHandler is None or Console is None:
|
||||
return False
|
||||
if os.getenv("NO_COLOR"):
|
||||
return False
|
||||
if os.getenv("TERM") == "dumb":
|
||||
return False
|
||||
return hasattr(sys.stderr, "isatty") and sys.stderr.isatty()
|
||||
|
||||
|
||||
def _allow_unicode() -> bool:
|
||||
encoding = (getattr(sys.stdout, "encoding", None) or "").lower()
|
||||
if not encoding:
|
||||
return False
|
||||
return "utf" in encoding
|
||||
|
||||
|
||||
def _console() -> Console:
|
||||
return Console(
|
||||
stderr=True,
|
||||
soft_wrap=False,
|
||||
highlight=False,
|
||||
force_terminal=_allow_color(),
|
||||
no_color=not _allow_color(),
|
||||
emoji=False,
|
||||
)
|
||||
|
||||
|
||||
def _rich_handler(level: str) -> RichHandler:
|
||||
return RichHandler(
|
||||
level=level,
|
||||
console=_console(),
|
||||
show_time=True,
|
||||
show_level=True,
|
||||
show_path=False,
|
||||
omit_repeated_times=False,
|
||||
markup=True,
|
||||
rich_tracebacks=True,
|
||||
tracebacks_show_locals=False,
|
||||
log_time_format="%H:%M:%S",
|
||||
)
|
||||
|
||||
|
||||
def _plain_handler(level: str) -> logging.StreamHandler:
|
||||
handler = logging.StreamHandler()
|
||||
handler.setLevel(level)
|
||||
handler.setFormatter(
|
||||
PlainFormatter("%(asctime)s | %(levelname)-7s | %(component)-10s | %(message)s")
|
||||
)
|
||||
return handler
|
||||
|
||||
|
||||
def _handler(level: str) -> logging.Handler:
|
||||
return _rich_handler(level) if _allow_color() else _plain_handler(level)
|
||||
|
||||
|
||||
def configure_logging(settings: LoggingSettings) -> None:
|
||||
level = settings.log_level.upper()
|
||||
root = logging.getLogger()
|
||||
root.handlers.clear()
|
||||
root.setLevel(level)
|
||||
root.addHandler(_handler(level))
|
||||
|
||||
for name in ("uvicorn", "uvicorn.error", "fastapi"):
|
||||
logger = logging.getLogger(name)
|
||||
logger.handlers.clear()
|
||||
logger.setLevel(level)
|
||||
logger.propagate = True
|
||||
|
||||
access_logger = logging.getLogger("uvicorn.access")
|
||||
access_logger.handlers.clear()
|
||||
access_logger.propagate = False
|
||||
access_logger.disabled = True
|
||||
|
||||
|
||||
def get_logger(name: str) -> logging.LoggerAdapter[logging.Logger]:
|
||||
component = name.rsplit(".", 1)[-1]
|
||||
return logging.LoggerAdapter(logging.getLogger(name), {"component": component})
|
||||
|
||||
|
||||
def _icon(name: str) -> str:
|
||||
ascii_icons = {
|
||||
"app": "#",
|
||||
"info": "i",
|
||||
"success": "+",
|
||||
"warning": "!",
|
||||
"error": "x",
|
||||
"debug": ".",
|
||||
"section": "=",
|
||||
"url": ">",
|
||||
"shutdown": "-",
|
||||
}
|
||||
unicode_icons = {
|
||||
"app": "◆",
|
||||
"info": "ℹ",
|
||||
"success": "✓",
|
||||
"warning": "▲",
|
||||
"error": "✖",
|
||||
"debug": "•",
|
||||
"section": "─",
|
||||
"url": "↳",
|
||||
"shutdown": "◦",
|
||||
}
|
||||
icons = unicode_icons if _allow_unicode() else ascii_icons
|
||||
return icons[name]
|
||||
|
||||
|
||||
def _style(name: str) -> str:
|
||||
return {
|
||||
"info": "bold cyan",
|
||||
"success": "bold green",
|
||||
"warning": "bold yellow",
|
||||
"error": "bold red",
|
||||
"debug": "dim",
|
||||
"section": "bold bright_blue",
|
||||
"muted": "grey62",
|
||||
}[name]
|
||||
|
||||
|
||||
def section_heading(title: str) -> None:
|
||||
logger = get_logger("data_entry_app.section")
|
||||
if _allow_color():
|
||||
_console().rule(Text(f" {title.upper()} ", style=_style("section")))
|
||||
return
|
||||
logger.info("%s %s %s", _icon("section") * 10, title.upper(), _icon("section") * 10)
|
||||
|
||||
|
||||
def startup_banner(status: StartupStatus) -> None:
|
||||
logger = get_logger("data_entry_app.startup")
|
||||
if _allow_color():
|
||||
console = _console()
|
||||
table = Table.grid(expand=False)
|
||||
table.add_column(style="bold white", justify="left")
|
||||
table.add_column(style="white", justify="left")
|
||||
table.add_row("Environment", status.environment)
|
||||
table.add_row("Version", status.version)
|
||||
table.add_row("Host", status.host)
|
||||
table.add_row("Port", str(status.port))
|
||||
table.add_row("Database", status.database)
|
||||
table.add_row("Mode", status.mode)
|
||||
table.add_row("Started", status.started_at)
|
||||
|
||||
console.rule(Text(f" {status.app_name} ", style="bold white"))
|
||||
console.print(Text("Clean startup. Clear status. Ready.", style="italic cyan"))
|
||||
console.print(table)
|
||||
console.print()
|
||||
console.print(Text("App is running at:", style="bold white"))
|
||||
console.print(Text(f" Local: {status.local_url}", style="cyan"))
|
||||
console.print(Text(f" Network: {status.network_url}", style="cyan"))
|
||||
console.print()
|
||||
return
|
||||
|
||||
logger.info("%s %s", _icon("app"), "Startup banner")
|
||||
logger.info("App : %s", status.app_name)
|
||||
logger.info("Environment : %s", status.environment)
|
||||
logger.info("Version : %s", status.version)
|
||||
logger.info("Host : %s", status.host)
|
||||
logger.info("Port : %s", status.port)
|
||||
logger.info("Database : %s", status.database)
|
||||
logger.info("Mode : %s", status.mode)
|
||||
logger.info("Started : %s", status.started_at)
|
||||
logger.info("Local : %s", status.local_url)
|
||||
logger.info("Network : %s", status.network_url)
|
||||
|
||||
|
||||
def status_message(level: str, message: str, *args: object, logger_name: str = "data_entry_app.status") -> None:
|
||||
palette = {
|
||||
"debug": logging.DEBUG,
|
||||
"info": logging.INFO,
|
||||
"success": logging.INFO,
|
||||
"warning": logging.WARNING,
|
||||
"error": logging.ERROR,
|
||||
}
|
||||
labels = {
|
||||
"debug": f"[{_icon('debug')}]",
|
||||
"info": f"[{_icon('info')}]",
|
||||
"success": f"[{_icon('success')}]",
|
||||
"warning": f"[{_icon('warning')}]",
|
||||
"error": f"[{_icon('error')}]",
|
||||
}
|
||||
styles = {
|
||||
"debug": _style("debug"),
|
||||
"info": _style("info"),
|
||||
"success": _style("success"),
|
||||
"warning": _style("warning"),
|
||||
"error": _style("error"),
|
||||
}
|
||||
logger = get_logger(logger_name)
|
||||
rendered = message % args if args else message
|
||||
if _allow_color():
|
||||
logger.log(palette[level], f"[{styles[level]}]{labels[level]}[/] {rendered}")
|
||||
else:
|
||||
logger.log(palette[level], "%s %s", labels[level], rendered)
|
||||
|
||||
|
||||
def success(message: str, *args: object, logger_name: str = "data_entry_app.status") -> None:
|
||||
status_message("success", message, *args, logger_name=logger_name)
|
||||
|
||||
|
||||
def warning(message: str, *args: object, logger_name: str = "data_entry_app.status") -> None:
|
||||
status_message("warning", message, *args, logger_name=logger_name)
|
||||
|
||||
|
||||
def info(message: str, *args: object, logger_name: str = "data_entry_app.status") -> None:
|
||||
status_message("info", message, *args, logger_name=logger_name)
|
||||
|
||||
|
||||
def debug(message: str, *args: object, logger_name: str = "data_entry_app.status") -> None:
|
||||
status_message("debug", message, *args, logger_name=logger_name)
|
||||
|
||||
|
||||
def fatal(message: str, *args: object, exc_info: bool = False, logger_name: str = "data_entry_app.status") -> None:
|
||||
logger = get_logger(logger_name)
|
||||
rendered = message % args if args else message
|
||||
if _allow_color():
|
||||
logger.error(f"[{_style('error')}][{_icon('error')}][/] {rendered}", exc_info=exc_info)
|
||||
else:
|
||||
logger.error("[%s] %s", _icon("error"), rendered, exc_info=exc_info)
|
||||
|
||||
|
||||
def shutdown_summary(*, uptime_seconds: float, requests_served: int, host: str, port: int) -> None:
|
||||
section_heading("Shutdown")
|
||||
logger = get_logger("data_entry_app.shutdown")
|
||||
summary = f"Uptime {uptime_seconds:.1f}s | Requests {requests_served} | Endpoint http://{host}:{port}"
|
||||
if _allow_color():
|
||||
logger.info(f"[{_style('debug')}]{_icon('shutdown')}[/] {summary}")
|
||||
else:
|
||||
logger.info("%s %s", _icon("shutdown"), summary)
|
||||
|
||||
|
||||
def describe_database(url: str) -> str:
|
||||
if url.startswith("sqlite"):
|
||||
return "sqlite"
|
||||
if "postgresql" in url:
|
||||
return "postgresql"
|
||||
if "mysql" in url:
|
||||
return "mysql"
|
||||
return url.split(":", 1)[0]
|
||||
|
||||
|
||||
def sanitize_database_target(url: str) -> str:
|
||||
if url.startswith("sqlite:///"):
|
||||
return url.removeprefix("sqlite:///")
|
||||
if "@" in url:
|
||||
return url.split("@", 1)[1]
|
||||
return url
|
||||
|
||||
|
||||
def startup_status(settings: LoggingSettings) -> StartupStatus:
|
||||
host = settings.host
|
||||
local_host = "localhost" if host in {"0.0.0.0", "::"} else host
|
||||
timestamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %Z")
|
||||
return StartupStatus(
|
||||
app_name=settings.app_name,
|
||||
version=settings.version,
|
||||
environment=settings.app_env,
|
||||
host=settings.host,
|
||||
port=settings.port,
|
||||
database=f"{describe_database(settings.database_url)} ({sanitize_database_target(settings.database_url)})",
|
||||
mode="verbose" if settings.log_verbose else "normal",
|
||||
started_at=timestamp,
|
||||
local_url=f"http://{local_host}:{settings.port}",
|
||||
network_url=f"http://{host}:{settings.port}",
|
||||
)
|
||||
|
||||
|
||||
def route_summary(routes: Iterable[object]) -> tuple[int, list[str]]:
|
||||
lines: list[str] = []
|
||||
count = 0
|
||||
for route in routes:
|
||||
path = getattr(route, "path", None)
|
||||
methods = getattr(route, "methods", None)
|
||||
if not path or not methods:
|
||||
continue
|
||||
filtered_methods = sorted(method for method in methods if method not in {"HEAD", "OPTIONS"})
|
||||
if not filtered_methods:
|
||||
continue
|
||||
count += 1
|
||||
lines.append(f"{','.join(filtered_methods):<7} {path}")
|
||||
return count, lines
|
||||
|
||||
|
||||
def log_request(
|
||||
*,
|
||||
method: str,
|
||||
path: str,
|
||||
status_code: int,
|
||||
duration_ms: float,
|
||||
client: str,
|
||||
content_length: str | None,
|
||||
) -> None:
|
||||
level = "info"
|
||||
if status_code >= 500:
|
||||
level = "error"
|
||||
elif status_code >= 400:
|
||||
level = "warning"
|
||||
elif path == "/health":
|
||||
level = "debug"
|
||||
|
||||
message = (
|
||||
f"{method:<6} {status_code:>3} {duration_ms:>7.1f}ms "
|
||||
f"{path:<36} client={client}"
|
||||
)
|
||||
if content_length:
|
||||
message += f" bytes={content_length}"
|
||||
status_message(level, message, logger_name="data_entry_app.http")
|
||||
|
||||
|
||||
class RequestTimer:
|
||||
def __init__(self) -> None:
|
||||
self.started = time.perf_counter()
|
||||
|
||||
@property
|
||||
def elapsed_ms(self) -> float:
|
||||
return (time.perf_counter() - self.started) * 1000
|
||||
Reference in New Issue
Block a user