from __future__ import annotations import json import logging import os import sys import time from collections.abc import Mapping from http import HTTPStatus from logging.config import dictConfig, fileConfig from typing import Any, IO, TYPE_CHECKING if sys.version_info >= (3, 11): import tomllib else: import tomli as tomllib if TYPE_CHECKING: from .config import Config from .typing import ResponseSummary, WWWScope def _create_logger( name: str, target: logging.Logger | str | None, level: str | None, sys_default: IO, *, propagate: bool = True, ) -> logging.Logger | None: if isinstance(target, logging.Logger): return target if target: logger = logging.getLogger(name) logger.handlers = [ logging.StreamHandler(sys_default) if target == "-" else logging.FileHandler(target) ] logger.propagate = propagate formatter = logging.Formatter( "%(asctime)s [%(process)d] [%(levelname)s] %(message)s", "[%Y-%m-%d %H:%M:%S %z]", ) logger.handlers[0].setFormatter(formatter) if level is not None: logger.setLevel(logging.getLevelName(level.upper())) return logger else: return None class Logger: def __init__(self, config: Config) -> None: self.access_log_format = config.access_log_format self.access_logger = _create_logger( "hypercorn.access", config.accesslog, config.loglevel, sys.stdout, propagate=False, ) self.error_logger = _create_logger( "hypercorn.error", config.errorlog, config.loglevel, sys.stderr ) if config.logconfig is not None: if config.logconfig.startswith("json:"): with open(config.logconfig[5:]) as file_: dictConfig(json.load(file_)) elif config.logconfig.startswith("toml:"): with open(config.logconfig[5:], "rb") as file_: dictConfig(tomllib.load(file_)) else: log_config = { "__file__": config.logconfig, "here": os.path.dirname(config.logconfig), } fileConfig(config.logconfig, defaults=log_config, disable_existing_loggers=False) else: if config.logconfig_dict is not None: dictConfig(config.logconfig_dict) async def access( self, request: WWWScope, response: ResponseSummary, request_time: float ) -> None: if self.access_logger is not None: self.access_logger.info( self.access_log_format, self.atoms(request, response, request_time) ) async def critical(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.critical(message, *args, **kwargs) async def error(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.error(message, *args, **kwargs) async def warning(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.warning(message, *args, **kwargs) async def info(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.info(message, *args, **kwargs) async def debug(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.debug(message, *args, **kwargs) async def exception(self, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.exception(message, *args, **kwargs) async def log(self, level: int, message: str, *args: Any, **kwargs: Any) -> None: if self.error_logger is not None: self.error_logger.log(level, message, *args, **kwargs) def atoms( self, request: WWWScope, response: ResponseSummary | None, request_time: float ) -> Mapping[str, str]: """Create and return an access log atoms dictionary. This can be overidden and customised if desired. It should return a mapping between an access log format key and a value. """ return AccessLogAtoms(request, response, request_time) def __getattr__(self, name: str) -> Any: return getattr(self.error_logger, name) class AccessLogAtoms(dict): def __init__( self, request: WWWScope, response: ResponseSummary | None, request_time: float ) -> None: for name, value in request["headers"]: self[f"{{{name.decode('latin1').lower()}}}i"] = value.decode("latin1") for name, value in os.environ.items(): self[f"{{{name.lower()}}}e"] = value protocol = request.get("http_version", "ws") client = request.get("client") if client is None: remote_addr = None elif len(client) == 2: remote_addr = f"{client[0]}:{client[1]}" elif len(client) == 1: remote_addr = client[0] else: # make sure not to throw UnboundLocalError remote_addr = f"" if request["type"] == "http": method = request["method"] else: method = "GET" query_string = request["query_string"].decode() path_with_qs = request["path"] + ("?" + query_string if query_string else "") status_code = "-" status_phrase = "-" if response is not None: for name, value in response.get("headers", []): # type: ignore self[f"{{{name.decode('latin1').lower()}}}o"] = value.decode("latin1") # type: ignore # noqa: E501 status_code = str(response["status"]) try: status_phrase = HTTPStatus(response["status"]).phrase except ValueError: status_phrase = f"" self.update( { "h": remote_addr, "l": "-", "t": time.strftime("[%d/%b/%Y:%H:%M:%S %z]"), "r": f"{method} {request['path']} {protocol}", "R": f"{method} {path_with_qs} {protocol}", "s": status_code, "st": status_phrase, "S": request["scheme"], "m": method, "U": request["path"], "Uq": path_with_qs, "q": query_string, "H": protocol, "b": self["{Content-Length}o"], "B": self["{Content-Length}o"], "f": self["{Referer}i"], "a": self["{User-Agent}i"], "T": int(request_time), "D": int(request_time * 1_000_000), "L": f"{request_time:.6f}", "p": f"<{os.getpid()}>", } ) def __getitem__(self, key: str) -> str: try: if key.startswith("{"): return super().__getitem__(key.lower()) else: return super().__getitem__(key) except KeyError: return "-"