|
| 1 | +import logging |
| 2 | +import sys |
| 3 | +from typing import Any, Union |
| 4 | + |
| 5 | +from loguru import logger |
| 6 | + |
| 7 | +from {{cookiecutter.project_name}}.settings import settings |
| 8 | + |
| 9 | +{%- if cookiecutter.otlp_enabled == "True" %} |
| 10 | +from opentelemetry.trace import INVALID_SPAN, INVALID_SPAN_CONTEXT, get_current_span |
| 11 | +{%- endif %} |
| 12 | + |
| 13 | + |
| 14 | +class InterceptHandler(logging.Handler): |
| 15 | + """ |
| 16 | + Default handler from examples in loguru documentation. |
| 17 | +
|
| 18 | + This handler intercepts all log requests and |
| 19 | + passes them to loguru. |
| 20 | +
|
| 21 | + For more info see: |
| 22 | + https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging |
| 23 | + """ |
| 24 | + |
| 25 | + def emit(self, record: logging.LogRecord) -> None: |
| 26 | + """ |
| 27 | + Propagates logs to loguru. |
| 28 | +
|
| 29 | + :param record: record to log. |
| 30 | + """ |
| 31 | + try: |
| 32 | + level: Union[str, int] = logger.level(record.levelname).name |
| 33 | + except ValueError: |
| 34 | + level = record.levelno |
| 35 | + |
| 36 | + # Find caller from where originated the logged message |
| 37 | + frame, depth = logging.currentframe(), 2 |
| 38 | + while frame.f_code.co_filename == logging.__file__: |
| 39 | + frame = frame.f_back # type: ignore |
| 40 | + depth += 1 |
| 41 | + |
| 42 | + logger.opt(depth=depth, exception=record.exc_info).log( |
| 43 | + level, |
| 44 | + record.getMessage(), |
| 45 | + ) |
| 46 | + |
| 47 | +{%- if cookiecutter.otlp_enabled == "True" %} |
| 48 | + |
| 49 | +def record_formatter(record: dict[str, Any]) -> str: |
| 50 | + """ |
| 51 | + Formats the record. |
| 52 | +
|
| 53 | + This function formats message |
| 54 | + by adding extra trace information to the record. |
| 55 | +
|
| 56 | + :param record: record information. |
| 57 | + :return: format string. |
| 58 | + """ |
| 59 | + log_format = ( |
| 60 | + "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> " |
| 61 | + "| <level>{level: <8}</level> " |
| 62 | + "| <magenta>trace_id={extra[trace_id]}</magenta> " |
| 63 | + "| <blue>span_id={extra[span_id]}</blue> " |
| 64 | + "| <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> " |
| 65 | + "- <level>{message}</level>\n" |
| 66 | + ) |
| 67 | + |
| 68 | + span = get_current_span() |
| 69 | + record["extra"]["span_id"] = 0 |
| 70 | + record["extra"]["trace_id"] = 0 |
| 71 | + if span != INVALID_SPAN: |
| 72 | + span_context = span.get_span_context() |
| 73 | + if span_context != INVALID_SPAN_CONTEXT: |
| 74 | + record["extra"]["span_id"] = format(span_context.span_id, "016x") |
| 75 | + record["extra"]["trace_id"] = format(span_context.trace_id, "032x") |
| 76 | + |
| 77 | + if record["exception"]: |
| 78 | + log_format = f"{log_format}{{'{{'}}exception{{'}}'}}" |
| 79 | + |
| 80 | + return log_format |
| 81 | + |
| 82 | +{%- endif %} |
| 83 | + |
| 84 | +def configure_logging() -> None: |
| 85 | + """Configures logging.""" |
| 86 | + loggers = ( |
| 87 | + logging.getLogger(name) |
| 88 | + for name in logging.root.manager.loggerDict |
| 89 | + if name.startswith("uvicorn.") |
| 90 | + ) |
| 91 | + for uvicorn_logger in loggers: |
| 92 | + uvicorn_logger.handlers = [] |
| 93 | + |
| 94 | + # change handler for default uvicorn logger |
| 95 | + intercept_handler = InterceptHandler() |
| 96 | + logging.getLogger("uvicorn").handlers = [intercept_handler] |
| 97 | + logging.getLogger("uvicorn.access").handlers = [intercept_handler] |
| 98 | + |
| 99 | + # set logs output, level and format |
| 100 | + logger.remove() |
| 101 | + logger.add( |
| 102 | + sys.stdout, |
| 103 | + level=settings.log_level.value, |
| 104 | + {%- if cookiecutter.otlp_enabled == "True" %} |
| 105 | + format=record_formatter, # type: ignore |
| 106 | + {%- endif %} |
| 107 | + ) |
0 commit comments