1from __future__ import annotations
 2
 3import datetime
 4import logging
 5import sys
 6import traceback
 7from typing import Any
 8
 9from plain.logs import get_framework_logger
10
11# Module-level loggers
12log = get_framework_logger()
13access_log = get_framework_logger("plain.server.access")
14
15# Maps field names that come from request headers
16_HEADER_FIELDS = {
17    "user_agent": "USER-AGENT",
18    "referer": "REFERER",
19}
20
21
22def _get_header(req: Any, header_name: str) -> str:
23    """Look up a header value from the request's header list."""
24    for name, value in req.headers:
25        if name == header_name:
26            return value
27    return ""
28
29
30def log_access(
31    resp: Any,
32    req: Any,
33    request_time: datetime.timedelta,
34) -> None:
35    """Log an access entry for a completed request."""
36    if not access_log.handlers or not access_log.isEnabledFor(logging.INFO):
37        return
38
39    from plain.runtime import settings
40
41    status = resp.status
42    if isinstance(status, str):
43        status = status.split(None, 1)[0]
44
45    context: dict[str, Any] = {}
46
47    for field in settings.SERVER_ACCESS_LOG_FIELDS:
48        if field == "method":
49            context["method"] = req.method
50        elif field == "path":
51            context["path"] = req.path
52        elif field == "status":
53            context["status"] = int(status)
54        elif field == "duration_ms":
55            context["duration_ms"] = int(request_time.total_seconds() * 1000)
56        elif field == "size":
57            context["size"] = getattr(resp, "sent", None) or 0
58        elif field == "ip":
59            if isinstance(req.peer_addr, tuple):
60                context["ip"] = req.peer_addr[0]
61            elif isinstance(req.peer_addr, str):
62                context["ip"] = req.peer_addr
63            else:
64                context["ip"] = ""
65        elif field == "url":
66            if req.query:
67                context["url"] = f"{req.path}?{req.query}"
68            else:
69                context["url"] = req.path
70        elif field == "query":
71            context["query"] = req.query or ""
72        elif field == "protocol":
73            context["protocol"] = f"HTTP/{req.version[0]}.{req.version[1]}"
74        elif header_name := _HEADER_FIELDS.get(field):
75            context[field] = _get_header(req, header_name)
76
77    try:
78        access_log.info("Request", extra=context)
79    except Exception:
80        log.error(traceback.format_exc())
81
82
83def configure_access_log(*, enabled: bool, log_format: str) -> None:
84    """Configure the access logger.
85
86    Always writes to stdout (separate from the LOG_STREAM setting)
87    and can be disabled entirely via the enabled flag.
88    """
89    from plain.logs.configure import create_log_formatter
90
91    access_log.setLevel(logging.INFO)
92    access_log.handlers.clear()
93    access_log.propagate = False
94
95    if enabled:
96        handler = logging.StreamHandler(sys.stdout)
97        handler.setFormatter(create_log_formatter(log_format))
98        access_log.addHandler(handler)