v7.0.0

Author: ddc

pythonlogs/core/factory.py

@@ -1,4 +1,5 @@
 import atexit
+import dataclasses
 import logging
 import threading
 import time
@@ -10,7 +11,6 @@
 from pythonlogs.core.settings import get_log_settings
 from pythonlogs.size_rotating import SizeRotatingLog as _SizeRotatingLogImpl
 from pythonlogs.timed_rotating import TimedRotatingLog as _TimedRotatingLogImpl
-from typing import assert_never
 
 
 @dataclass
@@ -217,6 +217,48 @@ def get_memory_limits(cls) -> dict[str, int]:
         with cls._registry_lock:
             return {"max_loggers": cls._max_loggers, "ttl_seconds": cls._logger_ttl}
 
+    # Mapping of logger types to their implementation classes and accepted fields
+    _LOGGER_IMPL = {
+        LoggerType.BASIC: (
+            _BasicLogImpl,
+            {"level", "name", "encoding", "datefmt", "timezone", "showlocation"},
+        ),
+        LoggerType.SIZE_ROTATING: (
+            _SizeRotatingLogImpl,
+            {
+                "level",
+                "name",
+                "directory",
+                "filenames",
+                "maxmbytes",
+                "daystokeep",
+                "encoding",
+                "datefmt",
+                "timezone",
+                "streamhandler",
+                "showlocation",
+            },
+        ),
+        LoggerType.TIMED_ROTATING: (
+            _TimedRotatingLogImpl,
+            {
+                "level",
+                "name",
+                "directory",
+                "filenames",
+                "when",
+                "sufix",
+                "daystokeep",
+                "encoding",
+                "datefmt",
+                "timezone",
+                "streamhandler",
+                "showlocation",
+                "rotateatutc",
+            },
+        ),
+    }
+
     @staticmethod
     def create_logger(logger_type: LoggerType | str, config: LoggerConfig | None = None, **kwargs) -> logging.Logger:
         """
@@ -225,7 +267,7 @@ def create_logger(logger_type: LoggerType | str, config: LoggerConfig | None = N
         Args:
             logger_type: Type of logger to create (LoggerType enum or string)
             config: LoggerConfig object with logger parameters
-            **kwargs: Individual logger parameters (for backward compatibility)
+            **kwargs: Individual logger parameters (kwargs take precedence over config)
 
         Returns:
             Configured logger instance
@@ -242,159 +284,21 @@ def create_logger(logger_type: LoggerType | str, config: LoggerConfig | None = N
                     f"Invalid logger type: {logger_type}. Valid types: {[t.value for t in LoggerType]}"
                 ) from err
 
-        # Merge config and kwargs (kwargs take precedence for backward compatibility)
+        # Merge config and kwargs (kwargs take precedence)
         if config is None:
             config = LoggerConfig()
-
-        # Create a new config with kwargs overriding config values
-        final_config = LoggerConfig(
-            level=kwargs.get("level", config.level),
-            name=kwargs.get("name", config.name),
-            directory=kwargs.get("directory", config.directory),
-            filenames=kwargs.get("filenames", config.filenames),
-            encoding=kwargs.get("encoding", config.encoding),
-            datefmt=kwargs.get("datefmt", config.datefmt),
-            timezone=kwargs.get("timezone", config.timezone),
-            streamhandler=kwargs.get("streamhandler", config.streamhandler),
-            showlocation=kwargs.get("showlocation", config.showlocation),
-            maxmbytes=kwargs.get("maxmbytes", config.maxmbytes),
-            when=kwargs.get("when", config.when),
-            sufix=kwargs.get("sufix", config.sufix),
-            rotateatutc=kwargs.get("rotateatutc", config.rotateatutc),
-            daystokeep=kwargs.get("daystokeep", config.daystokeep),
-        )
+        merged = {f.name: kwargs.get(f.name, getattr(config, f.name)) for f in dataclasses.fields(LoggerConfig)}
 
         # Convert enum values to strings for logger classes
-        level_str = final_config.level.value if isinstance(final_config.level, LogLevel) else final_config.level
-        when_str = final_config.when.value if isinstance(final_config.when, RotateWhen) else final_config.when
-
-        # Create logger based on type
-        match logger_type:
-            case LoggerType.BASIC:
-                return _BasicLogImpl(
-                    level=level_str,
-                    name=final_config.name,
-                    encoding=final_config.encoding,
-                    datefmt=final_config.datefmt,
-                    timezone=final_config.timezone,
-                    showlocation=final_config.showlocation,
-                ).init()
-            case LoggerType.SIZE_ROTATING:
-                return _SizeRotatingLogImpl(
-                    level=level_str,
-                    name=final_config.name,
-                    directory=final_config.directory,
-                    filenames=final_config.filenames,
-                    maxmbytes=final_config.maxmbytes,
-                    daystokeep=final_config.daystokeep,
-                    encoding=final_config.encoding,
-                    datefmt=final_config.datefmt,
-                    timezone=final_config.timezone,
-                    streamhandler=final_config.streamhandler,
-                    showlocation=final_config.showlocation,
-                ).init()
-            case LoggerType.TIMED_ROTATING:
-                return _TimedRotatingLogImpl(
-                    level=level_str,
-                    name=final_config.name,
-                    directory=final_config.directory,
-                    filenames=final_config.filenames,
-                    when=when_str,
-                    sufix=final_config.sufix,
-                    daystokeep=final_config.daystokeep,
-                    encoding=final_config.encoding,
-                    datefmt=final_config.datefmt,
-                    timezone=final_config.timezone,
-                    streamhandler=final_config.streamhandler,
-                    showlocation=final_config.showlocation,
-                    rotateatutc=final_config.rotateatutc,
-                ).init()
-            case _ as unreachable:  # pragma: no cover
-                assert_never(unreachable)
-
-    @staticmethod
-    def create_basic_logger(
-        level: LogLevel | str | None = None,
-        name: str | None = None,
-        encoding: str | None = None,
-        datefmt: str | None = None,
-        timezone: str | None = None,
-        showlocation: bool | None = None,
-    ) -> logging.Logger:
-        """Convenience method for creating a basic logger"""
-        return LoggerFactory.create_logger(
-            LoggerType.BASIC,
-            level=level,
-            name=name,
-            encoding=encoding,
-            datefmt=datefmt,
-            timezone=timezone,
-            showlocation=showlocation,
-        )
-
-    @staticmethod
-    def create_size_rotating_logger(
-        level: LogLevel | str | None = None,
-        name: str | None = None,
-        directory: str | None = None,
-        filenames: list | tuple | None = None,
-        maxmbytes: int | None = None,
-        daystokeep: int | None = None,
-        encoding: str | None = None,
-        datefmt: str | None = None,
-        timezone: str | None = None,
-        streamhandler: bool | None = None,
-        showlocation: bool | None = None,
-    ) -> logging.Logger:
-        """Convenience method for creating a size rotating logger"""
-        return LoggerFactory.create_logger(
-            LoggerType.SIZE_ROTATING,
-            level=level,
-            name=name,
-            directory=directory,
-            filenames=filenames,
-            maxmbytes=maxmbytes,
-            daystokeep=daystokeep,
-            encoding=encoding,
-            datefmt=datefmt,
-            timezone=timezone,
-            streamhandler=streamhandler,
-            showlocation=showlocation,
-        )
+        if isinstance(merged.get("level"), LogLevel):
+            merged["level"] = merged["level"].value
+        if isinstance(merged.get("when"), RotateWhen):
+            merged["when"] = merged["when"].value
 
-    @staticmethod
-    def create_timed_rotating_logger(
-        level: LogLevel | str | None = None,
-        name: str | None = None,
-        directory: str | None = None,
-        filenames: list | tuple | None = None,
-        when: RotateWhen | str | None = None,
-        sufix: str | None = None,
-        daystokeep: int | None = None,
-        encoding: str | None = None,
-        datefmt: str | None = None,
-        timezone: str | None = None,
-        streamhandler: bool | None = None,
-        showlocation: bool | None = None,
-        rotateatutc: bool | None = None,
-    ) -> logging.Logger:
-        """Convenience method for creating a timed rotating logger"""
-        return LoggerFactory.create_logger(
-            LoggerType.TIMED_ROTATING,
-            level=level,
-            name=name,
-            directory=directory,
-            filenames=filenames,
-            when=when,
-            sufix=sufix,
-            daystokeep=daystokeep,
-            encoding=encoding,
-            datefmt=datefmt,
-            timezone=timezone,
-            streamhandler=streamhandler,
-            showlocation=showlocation,
-            rotateatutc=rotateatutc,
-        )
+        # Create logger using table-driven dispatch
+        impl_class, valid_fields = LoggerFactory._LOGGER_IMPL[logger_type]
+        logger_kwargs = {k: v for k, v in merged.items() if k in valid_fields}
+        return impl_class(**logger_kwargs).init()
 
 
 # Public API wrapper classes - act like logging.Logger with context manager support
@@ -439,7 +343,8 @@ def __init__(
         timezone: str | None = None,
         showlocation: bool | None = None,
     ):
-        self._logger = LoggerFactory.create_basic_logger(
+        self._logger = LoggerFactory.create_logger(
+            LoggerType.BASIC,
             level=level,
             name=name,
             encoding=encoding,
@@ -477,7 +382,8 @@ def __init__(
         streamhandler: bool | None = None,
         showlocation: bool | None = None,
     ):
-        self._logger = LoggerFactory.create_size_rotating_logger(
+        self._logger = LoggerFactory.create_logger(
+            LoggerType.SIZE_ROTATING,
             level=level,
             name=name,
             directory=directory,
@@ -522,7 +428,8 @@ def __init__(
         showlocation: bool | None = None,
         rotateatutc: bool | None = None,
     ):
-        self._logger = LoggerFactory.create_timed_rotating_logger(
+        self._logger = LoggerFactory.create_logger(
+            LoggerType.TIMED_ROTATING,
             level=level,
             name=name,
             directory=directory,

tests/context_management/test_resource_management.py

@@ -51,7 +51,8 @@ def test_factory_registry_cleanup(self):
         logger_name = "test_registry_cleanup"
 
         # Create logger through factory (returns logging.Logger)
-        logger = LoggerFactory.create_size_rotating_logger(
+        logger = LoggerFactory.create_logger(
+            LoggerType.SIZE_ROTATING,
             name=logger_name,
             directory=self.temp_dir,
             filenames=[self.log_file],
@@ -150,7 +151,8 @@ def test_registry_clear_with_file_handlers(self):
         logger_name = "test_file_handlers"
 
         # Create logger with file handlers (returns logging.Logger)
-        logger = LoggerFactory.create_size_rotating_logger(
+        logger = LoggerFactory.create_logger(
+            LoggerType.SIZE_ROTATING,
             name=logger_name,
             directory=self.temp_dir,
             filenames=[self.log_file, "second.log"],
@@ -186,7 +188,8 @@ def test_resource_cleanup_performance(self):
         # Create multiple loggers using factory
         start_time = time.time()
         for name in logger_names:
-            logger = LoggerFactory.create_size_rotating_logger(
+            logger = LoggerFactory.create_logger(
+                LoggerType.SIZE_ROTATING,
                 name=name,
                 directory=self.temp_dir,
                 filenames=[f"{name}.log"],
@@ -218,7 +221,8 @@ def test_memory_usage_after_cleanup(self):
         logger_name = "memory_test_logger"
 
         # Create logger using factory (returns logging.Logger)
-        logger = LoggerFactory.create_size_rotating_logger(
+        logger = LoggerFactory.create_logger(
+            LoggerType.SIZE_ROTATING,
             name=logger_name,
             directory=self.temp_dir,
             filenames=[self.log_file],
@@ -258,7 +262,7 @@ def create_and_cleanup_logger(index):
             """Create a logger and immediately clean it up."""
             logger_name = f"concurrent_test_{index}"
             # Use factory to create logger (returns logging.Logger)
-            logger = LoggerFactory.create_basic_logger(name=logger_name, level=LogLevel.INFO.value)
+            logger = LoggerFactory.create_logger(LoggerType.BASIC, name=logger_name, level=LogLevel.INFO.value)
 
             # Add to registry
             LoggerFactory._logger_registry[logger_name] = (logger, time.time())

tests/factory/test_factory.py

@@ -164,8 +164,12 @@ def test_factory_create_logger_invalid_type(self):
     def test_factory_create_size_rotating_logger(self):
         """Test factory create_size_rotating_logger method."""
         with tempfile.TemporaryDirectory() as temp_dir:
-            logger = LoggerFactory.create_size_rotating_logger(
-                name="size_factory_test", directory=temp_dir, maxmbytes=10, level=LogLevel.INFO
+            logger = LoggerFactory.create_logger(
+                LoggerType.SIZE_ROTATING,
+                name="size_factory_test",
+                directory=temp_dir,
+                maxmbytes=10,
+                level=LogLevel.INFO,
             )
             assert logger.name == "size_factory_test"
 
@@ -176,14 +180,18 @@ def test_factory_create_size_rotating_logger(self):
     def test_factory_create_timed_rotating_logger(self):
         """Test factory create_timed_rotating_logger method."""
         with tempfile.TemporaryDirectory() as temp_dir:
-            logger = LoggerFactory.create_timed_rotating_logger(
-                name="timed_factory_test", directory=temp_dir, when=RotateWhen.DAILY, level="ERROR"
+            logger = LoggerFactory.create_logger(
+                LoggerType.TIMED_ROTATING,
+                name="timed_factory_test",
+                directory=temp_dir,
+                when=RotateWhen.DAILY,
+                level="ERROR",
             )
             assert logger.name == "timed_factory_test"
 
     def test_factory_create_basic_logger(self):
         """Test factory create_basic_logger method."""
-        logger = LoggerFactory.create_basic_logger(name="basic_factory_test", level=LogLevel.CRITICAL)
+        logger = LoggerFactory.create_logger(LoggerType.BASIC, name="basic_factory_test", level=LogLevel.CRITICAL)
         assert logger.name == "basic_factory_test"
         assert logger.level == 50  # CRITICAL level
 

tests/factory/test_factory_windows.py

@@ -69,7 +69,8 @@ def test_logger_with_file_output_windows(self):
     def test_factory_create_size_rotating_logger_windows(self):
         """Test factory create_size_rotating_logger method on Windows."""
         with windows_safe_temp_directory() as temp_dir:
-            logger = LoggerFactory.create_size_rotating_logger(
+            logger = LoggerFactory.create_logger(
+                LoggerType.SIZE_ROTATING,
                 name="size_factory_test_win",
                 directory=temp_dir,
                 maxmbytes=10,
@@ -81,7 +82,8 @@ def test_factory_create_size_rotating_logger_windows(self):
     def test_factory_create_timed_rotating_logger_windows(self):
         """Test factory create_timed_rotating_logger method on Windows."""
         with windows_safe_temp_directory() as temp_dir:
-            logger = LoggerFactory.create_timed_rotating_logger(
+            logger = LoggerFactory.create_logger(
+                LoggerType.TIMED_ROTATING,
                 name="timed_factory_test_win",
                 directory=temp_dir,
                 when=RotateWhen.DAILY,
@@ -379,15 +381,17 @@ def test_windows_file_locking_resilience_factory(self):
         """Test that Windows file locking resilience works with factory-created loggers."""
         with windows_safe_temp_directory() as temp_dir:
             # Create multiple loggers that write to the same directory
-            logger1 = LoggerFactory.create_size_rotating_logger(
+            logger1 = LoggerFactory.create_logger(
+                LoggerType.SIZE_ROTATING,
                 name="resilience_test1_win",
                 directory=temp_dir,
                 filenames=["resilience1.log"],
                 maxmbytes=1,
                 level=LogLevel.INFO,
             )
 
-            logger2 = LoggerFactory.create_timed_rotating_logger(
+            logger2 = LoggerFactory.create_logger(
+                LoggerType.TIMED_ROTATING,
                 name="resilience_test2_win",
                 directory=temp_dir,
                 filenames=["resilience2.log"],