DM-42273: Create new object, LogState

python/lsst/utils/logging.py

@@ -472,3 +472,62 @@ def log(self, msg: str, *args: Any) -> bool:
             self.num_issued += 1
             return True
         return False
+
+
+class LogState:
+    """Library-style helper to record and replay logging configuration.
+
+    This class stores a sequence of callable-and-arguments tuples that can be
+    serialized and replayed in a subprocess to reproduce the same logging
+    configuration that was applied in the parent process.
+
+    It is intentionally independent of CLI parsing logic; that is handled
+    by `CliLog`.
+    """
+
+    configState: list[tuple[Any, ...]] = []
+    _replayed: bool = False
+
+    @classmethod
+    def record(cls, value: tuple[Any, ...]) -> None:
+        """Append to configState contents"""
+        cls.configState.append(value)
+
+    @classmethod
+    def get_state(cls) -> list[tuple[Any, ...]]:
+        """Return a shallow copy of the current state."""
+        return list(cls.configState)
+
+    @classmethod
+    def set_state(cls, value: list[tuple[Any, ...]]) -> None:
+        """Replace configState contents"""
+        cls.configState.clear()
+        cls.configState.extend(value)
+
+    @classmethod
+    def clear_state(cls) -> None:
+        """Clear any recorded state."""
+        cls.configState.clear()
+        cls._replayed = False
+
+    @classmethod
+    def replay_state(cls, configState: list[tuple[Any, ...]]) -> None:
+        """Re-create configuration using configuration state recorded earlier.
+
+        Parameters
+        ----------
+        configState : `list` of `tuple`
+            Tuples contain a method as first item and arguments for the method.
+        """
+        if cls._replayed:
+            # Already initialized, do not touch anything.
+            log = logging.getLogger(__name__)
+            log.warning("Log is already initialized, will not replay configuration.")
+            return
+
+        # execute each one in order
+        for call in configState:
+            method, *args = call
+            method(*args)
+
+        cls._replayed = True

tests/test_logging.py

@@ -23,7 +23,7 @@
 import time
 import unittest
 
-from lsst.utils.logging import PeriodicLogger, getLogger, trace_set_at
+from lsst.utils.logging import LogState, PeriodicLogger, getLogger, trace_set_at
 
 
 class TestLogging(unittest.TestCase):
@@ -140,6 +140,67 @@ def test_periodic(self):
             periodic.log("Message")
         self.assertEqual(cm.records[0].filename, "test_logging.py", str(cm.records[0]))
 
+    def test_logstate(self):
+        class Myclass:
+            x: int = 0
+            y: float = 0.0
+            z: float = 0.0
+
+            @classmethod
+            def method1(cls, x: int) -> None:
+                cls.x = x
+
+            @classmethod
+            def method2(cls, y: float) -> None:
+                cls.y = y
+
+        # set the initial state
+        LogState.record((Myclass.method1, 451))
+        LogState.record((Myclass.method2, 3.14))
+
+        # check to see how it was set
+        state = LogState.get_state()
+        first = state[0]
+        self.assertEqual(first[0].__func__, Myclass.method1.__func__)
+        self.assertEqual(first[1], 451)
+        second = state[1]
+        self.assertEqual(second[0].__func__, Myclass.method2.__func__)
+        self.assertEqual(second[1], 3.14)
+
+        # ask state to be cleared
+        LogState.clear_state()
+
+        # make sure it was cleared
+        clear_state = LogState.get_state()
+        self.assertEqual(len(clear_state), 0)
+
+        # replay the state
+        LogState.replay_state(state)
+
+        # check to be sure the state was set as requested
+        self.assertEqual(Myclass.x, 451)
+        self.assertEqual(Myclass.y, 3.14)
+
+        # try to do it again, and receive a log warning
+        LogState.replay_state(state)
+
+        # clear the state; be sure it's cleared
+        LogState.clear_state()
+        clear_state = LogState.get_state()
+        self.assertEqual(len(clear_state), 0)
+
+        # call set state
+        LogState.set_state(state)
+
+        # and make sure it was set as requested
+        state = LogState.get_state()
+        first = state[0]
+        self.assertEqual(first[0].__func__, Myclass.method1.__func__)
+        self.assertEqual(first[1], 451)
+        second = state[1]
+        self.assertEqual(second[0].__func__, Myclass.method2.__func__)
+        self.assertEqual(second[1], 3.14)
+
 
 if __name__ == "__main__":
     unittest.main()