Skip to content

Logger

File: base/log/logger.py

This file implements the basic logger class that is used by the LOG functions.

Classes:

  • Logger –

    A thread-safe logger class that manages log handlers and provides logging functionality.

Logger

A thread-safe logger class that manages log handlers and provides logging functionality.

Attributes:

  • _instance_mutex –

    A mutex to ensure thread-safe access to the default logger instance.

  • _default_instance (Optional[Logger]) –

    The default logger instance, shared across threads.

  • _thread_local –

    Thread-local storage for logger instances.

  • _handlers_registry (dict[str, list[ILogHandler]]) –

    A registry of handlers for each logger.

  • _registry_mutex –

    A lock to ensure thread-safe access to the handler's registry.

Methods:

  • __init__ –

    Constructor.

  • add_handler –

    Adds a log handler to the logger.

  • remove_handler –

    Removes a log handler from the logger.

  • close_all_handlers –

    Closes all log handlers.

  • log –

    Logs a message with the specified severity level.

  • get_default –

    Retrieves the default logger instance, creating it if necessary.

  • get_thread_logger –

    Retrieves a thread-specific logger instance, creating it if necessary.

  • cleanup_thread_logger –

    Cleans up the thread-local logger instance, removing it from the registry.

Source code in kern_comm_lib/base/log/logger.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
class Logger:
  """A thread-safe logger class that manages log handlers and provides logging functionality.

  Attributes:
    _instance_mutex: A mutex to ensure thread-safe access to the default logger instance.
    _default_instance: The default logger instance, shared across threads.
    _thread_local: Thread-local storage for logger instances.
    _handlers_registry: A registry of handlers for each logger.
    _registry_mutex: A lock to ensure thread-safe access to the handler's registry.
  """

  _instance_mutex = mutex.MutexFactory.create_mutex()
  _default_instance: Optional["Logger"] = None
  _thread_local = threading.local()
  _handlers_registry: dict[str, list["ILogHandler"]] = {}
  _registry_mutex = mutex.MutexFactory.create_mutex()

  def __init__(self, name: str = "default") -> None:
    """Constructor.

    Args:
      name (default: "default"): The name of the logger.
    """
    self.name = name
    self._handlers: list[ILogHandler] = []
    self._handlers_lock = mutex.MutexFactory.create_mutex()

    # Register the logger's handlers in a thread-safe manner to the global
    # default instance
    with Logger._registry_mutex:
      Logger._handlers_registry[name] = self._handlers

  def add_handler(self, handler: "ILogHandler") -> Status:
    """Adds a log handler to the logger.

    Args:
      handler: The log handler to add.

    Returns:
      A Status object indicating success or failure.
    """
    # <editor-fold desc="Checks">
    DCHECK_NOT_NONE(handler)
    DCHECK_IS_TYPE(handler, ILogHandler)
    # </editor-fold>

    try:
      # The context manager ensures that the mutex is released even if an exception occurs
      with self._handlers_lock:
        if handler not in self._handlers:
          self._handlers.append(handler)
      return Status()
    except Exception as e:
      return Status.from_exception(e)

  def remove_handler(self, handler: "ILogHandler") -> Status:
    """Removes a log handler from the logger.

    Args:
      handler: The log handler to remove.

    Returns:
      A Status object indicating success or failure.
    """
    # <editor-fold desc="Checks">
    DCHECK_NOT_NONE(handler)
    DCHECK_IS_TYPE(handler, ILogHandler)
    # </editor-fold>

    try:
      # The context manager ensures that the mutex is released even if an exception occurs
      with self._handlers_lock:
        if handler in self._handlers:
          self._handlers.remove(handler)
      return Status()
    except Exception as e:
      return Status.from_exception(e)

  def close_all_handlers(self) -> Status:
    """Closes all log handlers.

    Returns:
      A Status object indicating success or failure of the logging operation.
    """
    try:
      with self._handlers_lock:
        for handler in self._handlers:
          tmp_status: Status = handler.close()
          if not tmp_status.ok():
            return tmp_status
        self._handlers.clear()
      return Status()
    except Exception as e:
      return Status.from_exception(e)

  def log(self, severity: "LogSeverity", message: str) -> Status:
    """Logs a message with the specified severity level.

    Args:
      severity: The severity level of the log message.
      message: The log message to be recorded.

    Returns:
      A Status object indicating success or failure of the logging operation.
    """
    # <editor-fold desc="Checks">
    DCHECK_NOT_NONE(severity)
    DCHECK_NOT_NONE(message)
    # </editor-fold>

    # Get a snapshot of handlers to prevent modification during iteration
    try:
      with self._handlers_lock:
        if not self._handlers:
          return Status.from_status_code(
              StatusCode.NOT_FOUND, "No handlers configured"
          )
        # The shallow copy of the _handlers list is sufficient because:
        # We only need to protect against list modification (add/remove handlers)
        # The handler objects themselves are expected to handle their own thread safety
        handlers = self._handlers.copy()
    except Exception as e:
      return Status.from_exception(e)

    # Process handlers outside the lock
    try:
      for handler in handlers:
        tmp_status: Status = handler.handle(severity, message)
        if not tmp_status.ok():
          return tmp_status
      return Status()
    except Exception as e:
      return Status.from_exception(e)

  @classmethod
  def get_default(cls) -> "Logger":
    """Retrieves the default logger instance, creating it if necessary.

    Returns:
      The default logger instance.
    """
    with cls._instance_mutex:
      if cls._default_instance is None:
        cls._default_instance = cls("default")
      return cls._default_instance

  @classmethod
  def get_thread_logger(cls, name: str | None = None) -> "Logger":
    """Retrieves a thread-specific logger instance, creating it if necessary.

    Args:
      name (default: None): The name of the logger. If None, a name is generated based on the thread name.

    Returns:
      The thread-specific logger instance.
    """
    if not hasattr(cls._thread_local, "logger"):
      with cls._instance_mutex:
        thread_name = name or f"thread_{threading.current_thread().name}"
        cls._thread_local.logger = cls(thread_name)
    return cls._thread_local.logger

  @classmethod
  def cleanup_thread_logger(cls) -> None:
    """Cleans up the thread-local logger instance, removing it from the registry."""
    if hasattr(cls._thread_local, "logger"):
      with cls._registry_mutex:
        name = cls._thread_local.logger.name
        if name in cls._handlers_registry:
          del cls._handlers_registry[name]
      delattr(cls._thread_local, "logger")

__init__ 

__init__(name: str = 'default') -> None

Constructor.

Parameters:

  • name (default, default: 'default' ) –

    "default"): The name of the logger.

Source code in kern_comm_lib/base/log/logger.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, name: str = "default") -> None:
  """Constructor.

  Args:
    name (default: "default"): The name of the logger.
  """
  self.name = name
  self._handlers: list[ILogHandler] = []
  self._handlers_lock = mutex.MutexFactory.create_mutex()

  # Register the logger's handlers in a thread-safe manner to the global
  # default instance
  with Logger._registry_mutex:
    Logger._handlers_registry[name] = self._handlers

add_handler 

add_handler(handler: ILogHandler) -> Status

Adds a log handler to the logger.

Parameters:

Returns:

  • Status –

    A Status object indicating success or failure.

Source code in kern_comm_lib/base/log/logger.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def add_handler(self, handler: "ILogHandler") -> Status:
  """Adds a log handler to the logger.

  Args:
    handler: The log handler to add.

  Returns:
    A Status object indicating success or failure.
  """
  # <editor-fold desc="Checks">
  DCHECK_NOT_NONE(handler)
  DCHECK_IS_TYPE(handler, ILogHandler)
  # </editor-fold>

  try:
    # The context manager ensures that the mutex is released even if an exception occurs
    with self._handlers_lock:
      if handler not in self._handlers:
        self._handlers.append(handler)
    return Status()
  except Exception as e:
    return Status.from_exception(e)

remove_handler 

remove_handler(handler: ILogHandler) -> Status

Removes a log handler from the logger.

Parameters:

Returns:

  • Status –

    A Status object indicating success or failure.

Source code in kern_comm_lib/base/log/logger.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def remove_handler(self, handler: "ILogHandler") -> Status:
  """Removes a log handler from the logger.

  Args:
    handler: The log handler to remove.

  Returns:
    A Status object indicating success or failure.
  """
  # <editor-fold desc="Checks">
  DCHECK_NOT_NONE(handler)
  DCHECK_IS_TYPE(handler, ILogHandler)
  # </editor-fold>

  try:
    # The context manager ensures that the mutex is released even if an exception occurs
    with self._handlers_lock:
      if handler in self._handlers:
        self._handlers.remove(handler)
    return Status()
  except Exception as e:
    return Status.from_exception(e)

close_all_handlers 

close_all_handlers() -> Status

Closes all log handlers.

Returns:

  • Status –

    A Status object indicating success or failure of the logging operation.

Source code in kern_comm_lib/base/log/logger.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def close_all_handlers(self) -> Status:
  """Closes all log handlers.

  Returns:
    A Status object indicating success or failure of the logging operation.
  """
  try:
    with self._handlers_lock:
      for handler in self._handlers:
        tmp_status: Status = handler.close()
        if not tmp_status.ok():
          return tmp_status
      self._handlers.clear()
    return Status()
  except Exception as e:
    return Status.from_exception(e)

log 

log(severity: LogSeverity, message: str) -> Status

Logs a message with the specified severity level.

Parameters:

  • severity (LogSeverity) –

    The severity level of the log message.

  • message (str) –

    The log message to be recorded.

Returns:

  • Status –

    A Status object indicating success or failure of the logging operation.

Source code in kern_comm_lib/base/log/logger.py 
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
170
171
172
173
174
175
176
def log(self, severity: "LogSeverity", message: str) -> Status:
  """Logs a message with the specified severity level.

  Args:
    severity: The severity level of the log message.
    message: The log message to be recorded.

  Returns:
    A Status object indicating success or failure of the logging operation.
  """
  # <editor-fold desc="Checks">
  DCHECK_NOT_NONE(severity)
  DCHECK_NOT_NONE(message)
  # </editor-fold>

  # Get a snapshot of handlers to prevent modification during iteration
  try:
    with self._handlers_lock:
      if not self._handlers:
        return Status.from_status_code(
            StatusCode.NOT_FOUND, "No handlers configured"
        )
      # The shallow copy of the _handlers list is sufficient because:
      # We only need to protect against list modification (add/remove handlers)
      # The handler objects themselves are expected to handle their own thread safety
      handlers = self._handlers.copy()
  except Exception as e:
    return Status.from_exception(e)

  # Process handlers outside the lock
  try:
    for handler in handlers:
      tmp_status: Status = handler.handle(severity, message)
      if not tmp_status.ok():
        return tmp_status
    return Status()
  except Exception as e:
    return Status.from_exception(e)

get_default classmethod 

get_default() -> Logger

Retrieves the default logger instance, creating it if necessary.

Returns:

  • Logger –

    The default logger instance.

Source code in kern_comm_lib/base/log/logger.py 
178
179
180
181
182
183
184
185
186
187
188
@classmethod
def get_default(cls) -> "Logger":
  """Retrieves the default logger instance, creating it if necessary.

  Returns:
    The default logger instance.
  """
  with cls._instance_mutex:
    if cls._default_instance is None:
      cls._default_instance = cls("default")
    return cls._default_instance

get_thread_logger classmethod 

get_thread_logger(name: str | None = None) -> Logger

Retrieves a thread-specific logger instance, creating it if necessary.

Parameters:

  • name (default, default: None ) –

    None): The name of the logger. If None, a name is generated based on the thread name.

Returns:

  • Logger –

    The thread-specific logger instance.

Source code in kern_comm_lib/base/log/logger.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@classmethod
def get_thread_logger(cls, name: str | None = None) -> "Logger":
  """Retrieves a thread-specific logger instance, creating it if necessary.

  Args:
    name (default: None): The name of the logger. If None, a name is generated based on the thread name.

  Returns:
    The thread-specific logger instance.
  """
  if not hasattr(cls._thread_local, "logger"):
    with cls._instance_mutex:
      thread_name = name or f"thread_{threading.current_thread().name}"
      cls._thread_local.logger = cls(thread_name)
  return cls._thread_local.logger

cleanup_thread_logger classmethod 

cleanup_thread_logger() -> None

Cleans up the thread-local logger instance, removing it from the registry.

Source code in kern_comm_lib/base/log/logger.py 
206
207
208
209
210
211
212
213
214
@classmethod
def cleanup_thread_logger(cls) -> None:
  """Cleans up the thread-local logger instance, removing it from the registry."""
  if hasattr(cls._thread_local, "logger"):
    with cls._registry_mutex:
      name = cls._thread_local.logger.name
      if name in cls._handlers_registry:
        del cls._handlers_registry[name]
    delattr(cls._thread_local, "logger")