Skip to content

Logging

This file configures logging for the application.

Logging is essential for capturing and recording events and messages generated during the execution of the application. This file provides a centralized place to configure the logging settings, such as log levels, log formats, log destinations, and more.

Logging settings can be adjusted to meet the specific needs of your application, including the desired log levels and the output destinations, such as console, file, or external services.

InterceptHandler

Bases: logging.Handler

Default handler from examples in loguru documentation.

This handler intercepts all log requests and passes them to loguru.

For more info see: https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging

Source code in hestia/logging.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class InterceptHandler(logging.Handler):
    """
    Default handler from examples in loguru documentation.

    This handler intercepts all log requests and
    passes them to loguru.

    For more info see:
    https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging
    """

    def emit(self, record: logging.LogRecord) -> None:  # pragma: no cover
        """
        Propagates logs to loguru.

        :param record: record to log.
        """
        try:
            level: Union[str, int] = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # Find caller from where originated the logged message
        frame, depth = logging.currentframe(), 2
        while frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back  # type: ignore
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level,
            record.getMessage(),
        )

emit(record)

Propagates logs to loguru.

Parameters:

Name Type Description Default
record logging.LogRecord

record to log.

 required
Source code in hestia/logging.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def emit(self, record: logging.LogRecord) -> None:  # pragma: no cover
    """
    Propagates logs to loguru.

    :param record: record to log.
    """
    try:
        level: Union[str, int] = logger.level(record.levelname).name
    except ValueError:
        level = record.levelno

    # Find caller from where originated the logged message
    frame, depth = logging.currentframe(), 2
    while frame.f_code.co_filename == logging.__file__:
        frame = frame.f_back  # type: ignore
        depth += 1

    logger.opt(depth=depth, exception=record.exc_info).log(
        level,
        record.getMessage(),
    )

configure_logging()

Configures logging.

Source code in hestia/logging.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def configure_logging() -> None:  # pragma: no cover
    """Configures logging."""
    intercept_handler = InterceptHandler()

    logging.basicConfig(handlers=[intercept_handler], level=logging.NOTSET)

    for logger_name in logging.root.manager.loggerDict:
        if logger_name.startswith("uvicorn."):
            logging.getLogger(logger_name).handlers = []
        if logger_name.startswith("taskiq."):
            logging.getLogger(logger_name).root.handlers = [intercept_handler]

    # change handler for default uvicorn logger
    logging.getLogger("uvicorn").handlers = [intercept_handler]
    logging.getLogger("uvicorn.access").handlers = [intercept_handler]

    # set logs output, level and format
    logger.remove()
    logger.add(
        sys.stdout,
        level=settings.log_level.value,
        format=record_formatter,  # type: ignore
    )

record_formatter(record)

Formats the record.

This function formats message by adding extra trace information to the record.

Parameters:

Name Type Description Default
record dict[str, Any]

record information.

 required

Returns:

Type Description
str

format string.
Source code in hestia/logging.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def record_formatter(record: dict[str, Any]) -> str:  # pragma: no cover
    """
    Formats the record.

    This function formats message
    by adding extra trace information to the record.

    :param record: record information.
    :return: format string.
    """
    log_format = (
        "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> "
        "| <level>{level: <8}</level> "
        "| <magenta>trace_id={extra[trace_id]}</magenta> "
        "| <blue>span_id={extra[span_id]}</blue> "
        "| <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> "
        "- <level>{message}</level>\n"
    )

    span = get_current_span()
    record["extra"]["span_id"] = 0
    record["extra"]["trace_id"] = 0
    if span != INVALID_SPAN:
        span_context = span.get_span_context()
        if span_context != INVALID_SPAN_CONTEXT:
            record["extra"]["span_id"] = format(span_context.span_id, "016x")
            record["extra"]["trace_id"] = format(span_context.trace_id, "032x")

    if record["exception"]:
        log_format = f"{log_format}{{exception}}"

    return log_format