Skip to content

kamihi.base.logging ⚓︎

Logging configuration module.

This module provides functions to configure logging for the Kamihi framework.

License

MIT

Examples:

>>> from kamihi.base.logging import configure_logging
>>> from kamihi.base.config import LogSettings
>>> from loguru import logger
>>> settings = LogSettings()
>>> configure_logging(logger, settings)
>>> logger.info("This is an info message.")

Classes:

Name Description
MongoLogger

MongoDB command logger.

Functions:

Name Description
configure_logging

Configure logging for the module.

MongoLogger ⚓︎

MongoLogger(logger: Logger)

Bases: CommandListener

MongoDB command logger.

This class listens to MongoDB commands and logs them using the loguru logger.

Parameters:

Name Type Description Default

logger ⚓︎

 Logger

The loguru logger instance to use for logging.

 required

Initialize the MongoLogger.

Parameters:

Name Type Description Default

logger ⚓︎

 Logger

The loguru logger instance to use for logging.

 required

Methods:

Name Description
failed

Log the failure of a command.
started

Log the start of a command.
succeeded

Log the success of a command.
Source code in src/kamihi/base/logging.py 
43
44
45
46
47
48
49
50
51
52
def __init__(self, logger: loguru.Logger) -> None:
    """
    Initialize the MongoLogger.

    Args:
        logger: The loguru logger instance to use for logging.

    """
    super().__init__()
    self.logger = logger

failed ⚓︎

failed(event: CommandFailedEvent) -> None

Log the failure of a command.

Source code in src/kamihi/base/logging.py 
73
74
75
76
77
78
79
80
81
82
def failed(self, event: CommandFailedEvent) -> None:
    """Log the failure of a command."""
    self.logger.debug(
        "Request failed",
        command_name=event.command_name,
        request_id=event.request_id,
        connection_id=event.connection_id,
        micoseconds=event.duration_micros,
        error=event.failure,
    )

started ⚓︎

started(event: CommandStartedEvent) -> None

Log the start of a command.

Source code in src/kamihi/base/logging.py 
54
55
56
57
58
59
60
61
def started(self, event: monitoring.CommandStartedEvent) -> None:
    """Log the start of a command."""
    self.logger.trace(
        "Executing request",
        command_name=event.command_name,
        request_id=event.request_id,
        connection_id=event.connection_id,
    )

succeeded ⚓︎

succeeded(event: CommandSucceededEvent) -> None

Log the success of a command.

Source code in src/kamihi/base/logging.py 
63
64
65
66
67
68
69
70
71
def succeeded(self, event: monitoring.CommandSucceededEvent) -> None:
    """Log the success of a command."""
    self.logger.trace(
        "Request succeeded",
        command_name=event.command_name,
        request_id=event.request_id,
        connection_id=event.connection_id,
        micoseconds=event.duration_micros,
    )

configure_logging ⚓︎

configure_logging(
    logger: Logger, settings: LogSettings
) -> None

Configure logging for the module.

This function sets up the logging configuration for the module, including log level and format.

Parameters:

Name Type Description Default

logger ⚓︎

 Logger

The logger instance to configure.

 required

settings ⚓︎

 LogSettings

The logging settings to configure.

 required
Source code in src/kamihi/base/logging.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def configure_logging(logger: loguru.Logger, settings: LogSettings) -> None:
    """
    Configure logging for the module.

    This function sets up the logging configuration for the module, including
    log level and format.

    Args:
        logger: The logger instance to configure.
        settings: The logging settings to configure.

    """
    logger.remove()

    logger.configure(patcher=_extra_formatter, extra={"compact": ""})

    if settings.stdout_enable:
        logger.add(
            sys.stdout,
            level=settings.stdout_level,
            format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            "{message} "
            "<dim>{extra[compact]}</dim>",
            serialize=settings.stdout_serialize,
            enqueue=True,
        )

    if settings.stderr_enable:
        logger.add(
            sys.stderr,
            level=settings.stderr_level,
            format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            "{message} "
            "<dim>{extra[compact]}</dim>",
            serialize=settings.stderr_serialize,
            enqueue=True,
        )

    if settings.file_enable:
        logger.add(
            settings.file_path,
            level=settings.file_level,
            format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            "{message} "
            "<dim>{extra[compact]}</dim>",
            serialize=settings.file_serialize,
            rotation=settings.file_rotation,
            retention=settings.file_retention,
            enqueue=True,
        )

    if settings.notification_enable:
        manual_sender = ManualSender(settings.notification_urls)
        logger.add(
            manual_sender.notify,
            level=settings.notification_level,
            format="<green>{time:YYYY-MM-DD at HH:mm:ss}</green> | "
            "<level>{level: <8}</level> | "
            "{message} "
            "<dim>{extra[compact]}</dim>",
            filter={"apprise": False},
            enqueue=True,
        )

    monitoring.register(MongoLogger(logger))