logging.py 4.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117
  1. from __future__ import annotations
  2. import http
  3. import logging
  4. import sys
  5. from copy import copy
  6. from typing import Literal
  7. import click
  8. TRACE_LOG_LEVEL = 5
  9. class ColourizedFormatter(logging.Formatter):
  10. """
  11. A custom log formatter class that:
  12. * Outputs the LOG_LEVEL with an appropriate color.
  13. * If a log call includes an `extra={"color_message": ...}` it will be used
  14. for formatting the output, instead of the plain text message.
  15. """
  16. level_name_colors = {
  17. TRACE_LOG_LEVEL: lambda level_name: click.style(str(level_name), fg="blue"),
  18. logging.DEBUG: lambda level_name: click.style(str(level_name), fg="cyan"),
  19. logging.INFO: lambda level_name: click.style(str(level_name), fg="green"),
  20. logging.WARNING: lambda level_name: click.style(str(level_name), fg="yellow"),
  21. logging.ERROR: lambda level_name: click.style(str(level_name), fg="red"),
  22. logging.CRITICAL: lambda level_name: click.style(str(level_name), fg="bright_red"),
  23. }
  24. def __init__(
  25. self,
  26. fmt: str | None = None,
  27. datefmt: str | None = None,
  28. style: Literal["%", "{", "$"] = "%",
  29. use_colors: bool | None = None,
  30. ):
  31. if use_colors in (True, False):
  32. self.use_colors = use_colors
  33. else:
  34. self.use_colors = sys.stdout.isatty()
  35. super().__init__(fmt=fmt, datefmt=datefmt, style=style)
  36. def color_level_name(self, level_name: str, level_no: int) -> str:
  37. def default(level_name: str) -> str:
  38. return str(level_name) # pragma: no cover
  39. func = self.level_name_colors.get(level_no, default)
  40. return func(level_name)
  41. def should_use_colors(self) -> bool:
  42. return True # pragma: no cover
  43. def formatMessage(self, record: logging.LogRecord) -> str:
  44. recordcopy = copy(record)
  45. levelname = recordcopy.levelname
  46. seperator = " " * (8 - len(recordcopy.levelname))
  47. if self.use_colors:
  48. levelname = self.color_level_name(levelname, recordcopy.levelno)
  49. if "color_message" in recordcopy.__dict__:
  50. recordcopy.msg = recordcopy.__dict__["color_message"]
  51. recordcopy.__dict__["message"] = recordcopy.getMessage()
  52. recordcopy.__dict__["levelprefix"] = levelname + ":" + seperator
  53. return super().formatMessage(recordcopy)
  54. class DefaultFormatter(ColourizedFormatter):
  55. def should_use_colors(self) -> bool:
  56. return sys.stderr.isatty() # pragma: no cover
  57. class AccessFormatter(ColourizedFormatter):
  58. status_code_colours = {
  59. 1: lambda code: click.style(str(code), fg="bright_white"),
  60. 2: lambda code: click.style(str(code), fg="green"),
  61. 3: lambda code: click.style(str(code), fg="yellow"),
  62. 4: lambda code: click.style(str(code), fg="red"),
  63. 5: lambda code: click.style(str(code), fg="bright_red"),
  64. }
  65. def get_status_code(self, status_code: int) -> str:
  66. try:
  67. status_phrase = http.HTTPStatus(status_code).phrase
  68. except ValueError:
  69. status_phrase = ""
  70. status_and_phrase = f"{status_code} {status_phrase}"
  71. if self.use_colors:
  72. def default(code: int) -> str:
  73. return status_and_phrase # pragma: no cover
  74. func = self.status_code_colours.get(status_code // 100, default)
  75. return func(status_and_phrase)
  76. return status_and_phrase
  77. def formatMessage(self, record: logging.LogRecord) -> str:
  78. recordcopy = copy(record)
  79. (
  80. client_addr,
  81. method,
  82. full_path,
  83. http_version,
  84. status_code,
  85. ) = recordcopy.args # type: ignore[misc]
  86. status_code = self.get_status_code(int(status_code)) # type: ignore[arg-type]
  87. request_line = f"{method} {full_path} HTTP/{http_version}"
  88. if self.use_colors:
  89. request_line = click.style(request_line, bold=True)
  90. recordcopy.__dict__.update(
  91. {
  92. "client_addr": client_addr,
  93. "request_line": request_line,
  94. "status_code": status_code,
  95. }
  96. )
  97. return super().formatMessage(recordcopy)