Skip to content

Commit

Permalink
Handle multiple logger case.
Browse files Browse the repository at this point in the history
Improve implementation by ensuring an exception is logged by all
instantiated loggers.

This is achieved by creating a module-level LoggerRegistry class that
keeps track of all instantiated loggers and defines a `log_exception`
class method which is assigned to `sys.excepthook`. The class method
documents the exception to all registered logger objects in the form
of a warning.
  • Loading branch information
ioannis-vm committed Oct 31, 2024
1 parent 3217096 commit b51d532
Show file tree
Hide file tree
Showing 2 changed files with 68 additions and 16 deletions.
46 changes: 42 additions & 4 deletions pelicun/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
import traceback
import warnings
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, TypeVar, overload
from typing import TYPE_CHECKING, Any, ClassVar, Optional, TypeVar, overload

import colorama
import numpy as np
Expand Down Expand Up @@ -253,6 +253,44 @@ def rng(self) -> np.random.Generator:
return self._rng


# Define a module-level LoggerRegistry
class LoggerRegistry:
"""Registry to manage all logger instances."""

_loggers: ClassVar[list[Logger]] = []

# The @classmethod decorator allows this method to be called on
# the class itself, rather than on instances. It interacts with
# class-level data (like _loggers), enabling a single registry for
# all Logger instances without needing an object of LoggerRegistry
# itself.
@classmethod
def register(cls, logger: Logger) -> None:
"""Register a logger instance."""
cls._loggers.append(logger)

@classmethod
def log_exception(
cls,
exc_type: type[BaseException],
exc_value: BaseException,
exc_traceback: TracebackType | None,
) -> None:
"""Log exceptions to all registered loggers."""
message = (
f"Unhandled exception occurred:"
f"\n"
f"{''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))}"
)
for logger in cls._loggers:
logger.warning(message)


# Update sys.excepthook to log exceptions in all loggers
# https://docs.python.org/3/library/sys.html#sys.excepthook
sys.excepthook = LoggerRegistry.log_exception


class Logger:
"""Generate log files documenting execution events."""

Expand Down Expand Up @@ -330,9 +368,9 @@ def __init__(
self.reset_log_strings()
control_warnings()

# Set sys.excepthook to handle uncaught exceptions
# https://docs.python.org/3/library/sys.html#sys.excepthook
sys.excepthook = self.log_exception
# Register the logger to the LoggerRegistry in order to
# capture raised exceptions.
LoggerRegistry.register(self)

def reset_log_strings(self) -> None:
"""Populate the string-related attributes of the logger."""
Expand Down
38 changes: 26 additions & 12 deletions pelicun/tests/basic/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -221,16 +221,26 @@ def test_logger_exception() -> None:
# Create a sample Python script that will raise an exception
test_script = Path(temp_dir) / 'test_script.py'
test_script_content = f"""
import sys
import traceback
from pathlib import Path
from pelicun.base import Logger
log_file = "{Path(temp_dir) / 'log.txt'}"
log_file_A = Path("{temp_dir}") / 'log_A.txt'
log_file_B = Path("{temp_dir}") / 'log_B.txt'
log = Logger(log_file=log_file, verbose=True, log_show_ms=True, print_log=True)
log_A = Logger(
log_file=log_file_A,
verbose=True,
log_show_ms=True,
print_log=True,
)
log_B = Logger(
log_file=log_file_B,
verbose=True,
log_show_ms=True,
print_log=True,
)
raise ValueError("Test exception in subprocess")
raise ValueError('Test exception in subprocess')
"""

# Write the test script to the file
Expand All @@ -248,15 +258,19 @@ def test_logger_exception() -> None:
assert process.returncode == 1

# Check the stdout/stderr for the expected output
assert 'Test exception in subprocess' in process.stderr
assert 'Test exception in subprocess' in process.stdout

# Check that the exception was logged in the log file
log_file = Path(temp_dir) / 'log.txt'
assert log_file.exists(), 'Log file was not created'
log_content = log_file.read_text()
assert 'Test exception in subprocess' in log_content
assert 'Traceback' in log_content
assert 'ValueError' in log_content
log_files = (
Path(temp_dir) / 'log_A_warnings.txt',
Path(temp_dir) / 'log_B_warnings.txt',
)
for log_file in log_files:
assert log_file.exists(), 'Log file was not created'
log_content = log_file.read_text()
assert 'Test exception in subprocess' in log_content
assert 'Traceback' in log_content
assert 'ValueError' in log_content


def test_split_file_name() -> None:
Expand Down

0 comments on commit b51d532

Please sign in to comment.