Skip to content

Log handlers

File: base/log/log_handlers.py

This file implements two log handlers: - FileLogHandler: Writes log messages to a file. - ConsoleLogHandler: Writes log messages to the console with color formatting.

Classes:

ILogHandler

Bases: ABC

Abstract base class for log handlers.

Methods:

  • handle –

    Handles a log message.

  • close –

    Closes the log handler.

Source code in kern_comm_lib/base/log/log_handlers.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class ILogHandler(ABC):
  """Abstract base class for log handlers."""

  @abstractmethod
  def handle(
      self, severity: "log_severity.LogSeverity", message: str
  ) -> Status:
    """Handles a log message.

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

    Returns:
      A Status object indicating success or failure of the operation.
    """
    pass

  def close(self) -> Status:
    """Closes the log handler.

    This method should be overridden by subclasses to implement any necessary
    cleanup operations when the handler is no longer needed.
    """
    return Status()

handle abstractmethod 

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

Handles a log message.

Parameters:

  • severity (LogSeverity) –

    The severity level of the log message.

  • message (str) –

    The log message to be handled.

Returns:

  • Status –

    A Status object indicating success or failure of the operation.

Source code in kern_comm_lib/base/log/log_handlers.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
def handle(
    self, severity: "log_severity.LogSeverity", message: str
) -> Status:
  """Handles a log message.

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

  Returns:
    A Status object indicating success or failure of the operation.
  """
  pass

close 

close() -> Status

Closes the log handler.

This method should be overridden by subclasses to implement any necessary cleanup operations when the handler is no longer needed.

Source code in kern_comm_lib/base/log/log_handlers.py 
61
62
63
64
65
66
67
def close(self) -> Status:
  """Closes the log handler.

  This method should be overridden by subclasses to implement any necessary
  cleanup operations when the handler is no longer needed.
  """
  return Status()

FileLogHandler

Bases: ILogHandler

Log handler that writes log messages to a file.

Attributes:

  • file_path (str) –

    The path to the log file.

  • _mutex (IMutex) –

    A lock to ensure thread-safe file access.

  • file –

    The file object for the log file.

Methods:

  • __init__ –

    Initializes a FileLogHandler instance.

  • handle –

    Writes a log message to the file.

  • close –

    Closes the log file.

Source code in kern_comm_lib/base/log/log_handlers.py 
 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
class FileLogHandler(ILogHandler):
  """Log handler that writes log messages to a file.

  Attributes:
      file_path: The path to the log file.
      _mutex: A lock to ensure thread-safe file access.
      file: The file object for the log file.
  """

  def __init__(self, file_path: str, format_pattern: str | None = None) -> None:
    """Initializes a FileLogHandler instance.

    Args:
      file_path: The path to the log file.
    """
    self.file_path: str = file_path
    self._mutex: IMutex = MutexFactory.create_mutex()
    self._formatter: LogFormatter = LogFormatter(format_pattern)
    try:
      # self.file should contain the file object, therefore opening the
      # file_path using a context manager is not applicable here.
      self.file = open(self.file_path, "a")  # noqa: SIM115 (see above)
    except Exception as e:
      print(f"Failed to open log file: {str(e)}")
      sys.exit(1)

  def handle(
      self, severity: "log_severity.LogSeverity", message: str
  ) -> Status:
    """Writes a log message to the file.

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

    Returns:
      A Status object indicating success or failure of the operation.
    """
    try:
      formatted_message = self._formatter.format(severity, message)
      with self._mutex:
        self.file.write(f"{formatted_message}\n")
        self.file.flush()
      return Status()
    except Exception as e:
      return Status.from_exception(e)

  def close(self) -> Status:
    """Closes the log file.

    Ensures that the file is properly closed, even if an exception occurs.
    """
    with self._mutex:
      try:
        self.file.close()
        return Status()
      except Exception as e:
        return Status.from_exception(e)

__init__ 

__init__(file_path: str, format_pattern: str | None = None) -> None

Initializes a FileLogHandler instance.

Parameters:

  • file_path (str) –

    The path to the log file.

Source code in kern_comm_lib/base/log/log_handlers.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(self, file_path: str, format_pattern: str | None = None) -> None:
  """Initializes a FileLogHandler instance.

  Args:
    file_path: The path to the log file.
  """
  self.file_path: str = file_path
  self._mutex: IMutex = MutexFactory.create_mutex()
  self._formatter: LogFormatter = LogFormatter(format_pattern)
  try:
    # self.file should contain the file object, therefore opening the
    # file_path using a context manager is not applicable here.
    self.file = open(self.file_path, "a")  # noqa: SIM115 (see above)
  except Exception as e:
    print(f"Failed to open log file: {str(e)}")
    sys.exit(1)

handle 

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

Writes a log message to the file.

Parameters:

  • severity (LogSeverity) –

    The severity level of the log message.

  • message (str) –

    The log message to be written.

Returns:

  • Status –

    A Status object indicating success or failure of the operation.

Source code in kern_comm_lib/base/log/log_handlers.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def handle(
    self, severity: "log_severity.LogSeverity", message: str
) -> Status:
  """Writes a log message to the file.

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

  Returns:
    A Status object indicating success or failure of the operation.
  """
  try:
    formatted_message = self._formatter.format(severity, message)
    with self._mutex:
      self.file.write(f"{formatted_message}\n")
      self.file.flush()
    return Status()
  except Exception as e:
    return Status.from_exception(e)

close 

close() -> Status

Closes the log file.

Ensures that the file is properly closed, even if an exception occurs.

Source code in kern_comm_lib/base/log/log_handlers.py 
117
118
119
120
121
122
123
124
125
126
127
def close(self) -> Status:
  """Closes the log file.

  Ensures that the file is properly closed, even if an exception occurs.
  """
  with self._mutex:
    try:
      self.file.close()
      return Status()
    except Exception as e:
      return Status.from_exception(e)

ConsoleLogHandler

Bases: ILogHandler

Log handler that writes log messages to the console.

Attributes:

  • _mutex (IMutex) –

    A lock to ensure thread-safe console access.

Methods:

  • __init__ –

    Constructor.

  • handle –

    Writes a log message to the console with color formatting.

Source code in kern_comm_lib/base/log/log_handlers.py 
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
class ConsoleLogHandler(ILogHandler):
  """Log handler that writes log messages to the console.

  Attributes:
    _mutex: A lock to ensure thread-safe console access.
  """

  def __init__(self, format_pattern: str | None = None) -> None:
    """Constructor.

    Args:
      format_pattern: Optional format string for log messages
    """
    self._mutex: IMutex = MutexFactory.create_mutex()
    self._formatter: LogFormatter = LogFormatter(format_pattern)

  def handle(
      self, severity: "log_severity.LogSeverity", message: str
  ) -> Status:
    """Writes a log message to the console with color formatting.

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

    Returns:
      A Status object indicating success or failure of the operation.
    """
    try:
      color_code = {
          log_severity.INFO: "\033[0m",  # Default color
          log_severity.WARNING: "\033[33m",  # Yellow
          log_severity.ERROR: "\033[31m",  # Red
          log_severity.FATAL: "\033[35m",  # Magenta
      }.get(severity, "\033[0m")

      reset_code = "\033[0m"  # Reset color
      formatted_message = self._formatter.format(severity, message)
      formatted_message_with_color = (
          f"{color_code}{formatted_message}{reset_code}\n"
      )

      with self._mutex:
        sys.stdout.write(formatted_message_with_color)
        sys.stdout.flush()
      return Status()
    except Exception as e:
      return Status.from_exception(e)

__init__ 

__init__(format_pattern: str | None = None) -> None

Constructor.

Parameters:

  • format_pattern (str | None, default: None ) –

    Optional format string for log messages

Source code in kern_comm_lib/base/log/log_handlers.py 
137
138
139
140
141
142
143
144
def __init__(self, format_pattern: str | None = None) -> None:
  """Constructor.

  Args:
    format_pattern: Optional format string for log messages
  """
  self._mutex: IMutex = MutexFactory.create_mutex()
  self._formatter: LogFormatter = LogFormatter(format_pattern)

handle 

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

Writes a log message to the console with color formatting.

Parameters:

  • severity (LogSeverity) –

    The severity level of the log message.

  • message (str) –

    The log message to be written.

Returns:

  • Status –

    A Status object indicating success or failure of the operation.

Source code in kern_comm_lib/base/log/log_handlers.py 
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
def handle(
    self, severity: "log_severity.LogSeverity", message: str
) -> Status:
  """Writes a log message to the console with color formatting.

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

  Returns:
    A Status object indicating success or failure of the operation.
  """
  try:
    color_code = {
        log_severity.INFO: "\033[0m",  # Default color
        log_severity.WARNING: "\033[33m",  # Yellow
        log_severity.ERROR: "\033[31m",  # Red
        log_severity.FATAL: "\033[35m",  # Magenta
    }.get(severity, "\033[0m")

    reset_code = "\033[0m"  # Reset color
    formatted_message = self._formatter.format(severity, message)
    formatted_message_with_color = (
        f"{color_code}{formatted_message}{reset_code}\n"
    )

    with self._mutex:
      sys.stdout.write(formatted_message_with_color)
      sys.stdout.flush()
    return Status()
  except Exception as e:
    return Status.from_exception(e)