feat: Allow per scanner logging config

The following setup is now possible when scanners are run in own
scripts:

        import gallia
        import gallia.log
        import gallia.command

        logger = gallia.log.get_logger("gallia")

        class Scanner1(gallia.command.Script):
            def main(self) -> None:
                logger.info(f"hi {self.__class__.__name__}")
                logger.error("error")
                logger.warning("warning")
                logger.notice("notice")
                logger.info("info")
                logger.debug("debug")
                logger.trace("trace")

        class Scanner2(gallia.command.Script):
            def main(self) -> None:
                logger.info(f"hi {self.__class__.__name__}")

        if __name__ == "__main__":

            # Each scanner sets up its own logging setup
            # with a logger called `gallia`.
            Scanner1().entry_point()
            Scanner2().entry_point()

            # Alternatively, a context manager can be used
            # for more fine grained control.
            with gallia.log.setup_logging(
                logger_name="gallia",
                stderr_level=gallia.log.Loglevel.DEBUG,
                logfile="test.log.zst",
            ) as h:
                Scanner1(logging_handler=h).entry_point()
                Scanner2(logging_handler=h).entry_point()

src/gallia/cli/gallia.py

@@ -25,9 +25,8 @@
 from gallia.plugins.plugin import CommandTree, load_commands, load_plugins
 from gallia.pydantic_argparse import ArgumentParser
 from gallia.pydantic_argparse import BaseCommand as PydanticBaseCommand
-from gallia.utils import get_log_level
 
-setup_logging(Loglevel.DEBUG)
+setup_logging("gallia", Loglevel.DEBUG)
 
 
 defaults = dict[type, dict[str, Any]]
@@ -123,7 +122,6 @@ def get_command(config: BaseCommandConfig) -> BaseCommand:
 def parse_and_run(
     commands: type[BaseCommand] | MutableMapping[str, CommandTree | type[BaseCommand]],
     auto_complete: bool = True,
-    setup_log: bool = True,
     top_level_options: Mapping[str, Callable[[], None]] | None = None,
     show_help_on_zero_args: bool = True,
 ) -> Never:
@@ -136,7 +134,6 @@ def parse_and_run(
 
     :param commands: A hierarchy of commands.
     :param auto_complete: Turns auto-complete functionality on.
-    :param setup_log: Setup logging according to the parameters in the parsed config.
     :param top_level_options: Optional top-level actions, such as "--version", given by a mapping of arguments and
                               functions. The program redirects control to the given function, once the program is
                               called with the corresponding argument and terminates after it returns.
@@ -182,12 +179,6 @@ def __call__(
 
     assert isinstance(config, BaseCommandConfig)
 
-    if setup_log:
-        setup_logging(
-            level=get_log_level(config.verbose),
-            no_volatile_info=not config.volatile_info,
-        )
-
     sys.exit(get_command(config).entry_point())
 
 

src/gallia/command/base.py

@@ -14,7 +14,6 @@
 from collections.abc import MutableMapping
 from datetime import UTC, datetime
 from enum import Enum, unique
-from logging import Handler
 from pathlib import Path
 from subprocess import CalledProcessError, run
 from tempfile import gettempdir
@@ -26,12 +25,12 @@
 from gallia.command.config import Field, GalliaBaseModel, Idempotent
 from gallia.db.handler import DBHandler
 from gallia.dumpcap import Dumpcap
-from gallia.log import add_zst_log_handler, get_logger, tz
+from gallia.log import LoggingSetupHandler, Loglevel, get_log_level, get_logger, setup_logging, tz
 from gallia.power_supply import PowerSupply
 from gallia.power_supply.uri import PowerSupplyURI
 from gallia.services.uds.core.exception import UDSException
 from gallia.transports import BaseTransport, TargetURI
-from gallia.utils import camel_to_snake, get_file_log_level
+from gallia.utils import camel_to_snake
 
 
 @unique
@@ -180,9 +179,11 @@ class BaseCommand(FlockMixin, ABC):
     #: a log message with level critical is logged.
     CATCHED_EXCEPTIONS: list[type[Exception]] = []
 
-    log_file_handlers: list[Handler]
-
-    def __init__(self, config: BaseCommandConfig) -> None:
+    def __init__(
+        self,
+        config: BaseCommandConfig = BaseCommandConfig(),
+        logging_handler: LoggingSetupHandler | None = None,
+    ) -> None:
         self.id = camel_to_snake(self.__class__.__name__)
         self.config = config
         self.artifacts_dir = Path()
@@ -195,7 +196,7 @@ def __init__(self, config: BaseCommandConfig) -> None:
         )
         self._lock_file_fd: int | None = None
         self.db_handler: DBHandler | None = None
-        self.log_file_handlers = []
+        self.provided_logging_handler = logging_handler
 
     @abstractmethod
     def run(self) -> int: ...
@@ -323,15 +324,25 @@ def entry_point(self) -> int:
 
         if self.HAS_ARTIFACTS_DIR:
             self.artifacts_dir = self.prepare_artifactsdir(
-                self.config.artifacts_base, self.config.artifacts_dir
+                self.config.artifacts_base,
+                self.config.artifacts_dir,
+            )
+
+        if self.provided_logging_handler is None:
+            stderr_level = get_log_level(self.config.verbose)
+            logging_handler = setup_logging(
+                logger_name="gallia",
+                stderr_level=stderr_level,
+                close_on_exit=False,
             )
-            self.log_file_handlers.append(
-                add_zst_log_handler(
+            if self.HAS_ARTIFACTS_DIR:
+                logging_handler.add_zst_file_handler(
                     logger_name="gallia",
                     filepath=self.artifacts_dir.joinpath(FileNames.LOGFILE.value),
-                    file_log_level=get_file_log_level(self.config),
+                    log_level=stderr_level if self.config.trace_log is False else Loglevel.TRACE,
                 )
-            )
+        else:
+            logging_handler = self.provided_logging_handler
 
         if self.config.hooks:
             self.run_hook(HookVariant.PRE)
@@ -380,6 +391,9 @@ def entry_point(self) -> int:
         if self._lock_file_fd is not None:
             self._release_flock()
 
+        if self.provided_logging_handler is None:
+            logging_handler.stop_logging()
+
         return exit_code