Target project no longer depends on httpx but depends on ~matching fastapi-vue version. Ruff cleanup.

This commit is contained in:
2026-08-31 17:07:49 +00:00
parent cb0b1f067d
commit c65d8eaa12
5 changed files with 73 additions and 35 deletions
+13 -7
View File
@@ -179,8 +179,8 @@ def _http_access_log_extra(
) -> dict[str, object]:
client_addr = _client_host(scope)
full_path = _path(scope)
method = method if method is not None else cast(str, scope.get("method", "-"))
method = cast(str, scope.get("state", {}).get("access_log_method") or method)
method = method if method is not None else cast("str", scope.get("method", "-"))
method = cast("str", scope.get("state", {}).get("access_log_method") or method)
try:
status_phrase = http.HTTPStatus(status).phrase
@@ -318,18 +318,21 @@ def _assemble_access_log(fields: dict[str, object]) -> str:
class AccessLogMiddleware:
"""ASGI middleware logging HTTP and WebSocket access with colored fields."""
def __init__(self, app: ASGI3Application) -> None:
"""Store the wrapped app."""
self.app = app
async def __call__(
self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
) -> None:
"""Dispatch by scope type to HTTP/WebSocket access logging."""
if scope["type"] == "http":
return await self._handle_http(scope, receive, send)
elif scope["type"] == "websocket":
if scope["type"] == "websocket":
return await self._handle_websocket(scope, receive, send)
else:
return await self.app(scope, receive, send)
return await self.app(scope, receive, send)
async def _handle_http(
self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
@@ -346,7 +349,10 @@ class AccessLogMiddleware:
extra=www_scope.get("state", {}).get("log_extra", ""),
)
logger.info(
f'{fields["client_addr"]} - "{fields["request_line"]}" {fields["status_code"]}',
'%s - "%s" %s',
fields["client_addr"],
fields["request_line"],
fields["status_code"],
extra=fields,
)
await send(message)
@@ -369,7 +375,7 @@ class AccessLogMiddleware:
def _close_fields(message: ASGIReceiveEvent | ASGISendEvent) -> dict[str, object]:
if accepted:
assert ws_id is not None
assert ws_id is not None # noqa: S101 # guaranteed once accepted
return _ws_close_extra(
www_scope,
ws_id,
+9 -4
View File
@@ -28,6 +28,7 @@ ACCESS_LOGGER = "fastapi_vue.access"
def strip_ansi(text: str) -> str:
"""Remove ANSI escape codes from text."""
return ANSI_ESCAPE_RE.sub("", text)
@@ -51,7 +52,7 @@ class Formatter(logging.Formatter):
Records with the middleware's access fields (``client`` etc.) are
formatted from those; anything else gets an emoji level prefix
(``LEVEL: `` fallback for unknown levels) in place of uvicorn's
``levelprefix``. ANSI codes are stripped when colors are disabled.
``levelprefix``.
Instantiation always loads tracerite, and with ``access=True`` also
installs the access-log middleware: ``dictConfig`` builds formatters while
@@ -65,10 +66,11 @@ class Formatter(logging.Formatter):
fmt: str | None = None,
datefmt: str | None = None,
style: Literal["%", "{", "$"] = "%",
use_colors: bool | None = None,
use_colors: bool | None = None, # noqa: FBT001 # mirrors logging.Formatter
*,
access: bool = False,
) -> None:
"""Load tracerite, optionally install the access log, detect color support."""
tracerite.load()
if access:
install_access_log()
@@ -78,7 +80,8 @@ class Formatter(logging.Formatter):
self.use_colors = sys.stdout.isatty()
super().__init__(fmt=fmt, datefmt=datefmt, style=style)
def formatMessage(self, record: logging.LogRecord) -> str:
def formatMessage(self, record: logging.LogRecord) -> str: # noqa: N802
"""Format access records via middleware fields, others with an emoji prefix."""
if "client" not in record.__dict__:
return _level_prefix(record) + record.getMessage()
formatted = super().formatMessage(record)
@@ -98,6 +101,7 @@ class WebSocketChatterFilter(logging.Filter):
_PREFIXES = ('%s - "WebSocket ', "connection open", "connection closed", "connection rejected")
def filter(self, record: logging.LogRecord) -> bool:
"""Keep records not matching stock WebSocket chatter prefixes."""
msg = record.msg
if not isinstance(msg, str):
return True
@@ -113,6 +117,7 @@ class UvicornQuietFilter(logging.Filter):
"""
def filter(self, record: logging.LogRecord) -> bool:
"""Drop uvicorn records below WARNING."""
return not (record.name.startswith("uvicorn") and record.levelno < logging.WARNING)
@@ -125,7 +130,7 @@ def install_access_log() -> None:
The guard is deliberately module-level: reload/worker subprocesses
re-import this module, resetting it so the patch is re-applied there.
"""
global _installed
global _installed # noqa: PLW0603 # deliberately module-level, see docstring
if _installed:
return
_installed = True