Add support for (): 'class' syntax

Author: dolfinus

src/pydantic_settings_logging/__init__.py

@@ -11,7 +11,7 @@
 from pathlib import Path
 from typing import Any, Literal
 
-from pydantic import AliasChoices, BaseModel, Field, field_validator
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator
 from pydantic_settings import (
     BaseSettings,
     PydanticBaseSettingsSource,
@@ -40,10 +40,22 @@
 ]
 
 
+# https://github.com/python/cpython/blob/v3.13.9/Lib/logging/config.py#L480-L492
+class CallableFactoryConfig(BaseModel):
+    model_config = ConfigDict(extra="allow", populate_by_name=True)
+
+    callable: str = Field(
+        description="Custom callable",
+        validation_alias=AliasChoices("callable", "()"),
+        serialization_alias="()",
+    )
+
+
 # Formatter Models
+# https://github.com/python/cpython/blob/v3.13.9/Lib/logging/config.py#L688
 class FormatterConfig(BaseModel):
     """Configuration for a logging formatter."""
-    
+
     format: str | None = Field(
         default="%(levelname)s:%(name)s:%(message)s",
         description="Format string for log messages"
@@ -74,6 +86,7 @@ class FormatterConfig(BaseModel):
 
 
 # Filter Models
+# https://github.com/python/cpython/blob/v3.13.9/Lib/logging/config.py#L732
 class FilterConfig(BaseModel):
     """Configuration for a logging filter."""
 
@@ -90,11 +103,12 @@ class FilterConfig(BaseModel):
 
 
 # Handler Models
+# https://github.com/python/cpython/blob/v3.13.9/Lib/logging/config.py#L768
 class BaseHandlerConfig(BaseModel):
     """Base configuration for all handlers."""
-    
-    model_config = {"extra": "allow"}
-    
+
+    model_config = ConfigDict(extra="allow")
+
     class_: str = Field(
         validation_alias=AliasChoices("class_", "class"),
         serialization_alias="class",
@@ -389,7 +403,7 @@ class QueueHandlerConfig(BaseHandlerConfig):
         validation_alias=AliasChoices("class_", "class"),
         serialization_alias="class"
     )
-    queue: str = Field(
+    queue: str | CallableFactoryConfig = Field(
         description="Queue object reference"
     )
 
@@ -402,7 +416,7 @@ class QueueListenerConfig(BaseHandlerConfig):
         validation_alias=AliasChoices("class_", "class"),
         serialization_alias="class"
     )
-    queue: str = Field(
+    queue: str | CallableFactoryConfig = Field(
         description="Queue object reference"
     )
     handlers: list[str] = Field(
@@ -421,7 +435,7 @@ class QueueListenerConfig(BaseHandlerConfig):
 # Logger configuration
 class LoggerConfig(BaseModel):
     """Configuration for a logger."""
-    
+
     level: str | None = Field(
         default=None,
         description="Logging level"
@@ -478,7 +492,7 @@ def _load_file(self) -> dict[str, Any]:
 
         with open(self.toml_file, "rb") as f:
             data = tomllib.load(f)
-        
+
         # Navigate to specified table if provided
         if self.toml_table:
             for key in self.toml_table:
@@ -508,11 +522,11 @@ def __init__(
         super().__init__(settings_cls)
         self.json_file = json_file
         self._data = self._load_file()
-    
+
     def _load_file(self) -> dict[str, Any]:
         if not self.json_file or not Path(self.json_file).exists():
             return {}
-        
+
         with open(self.json_file, "r") as f:
             return json.load(f)
 
@@ -565,7 +579,7 @@ def _convert_ini_to_dictconfig(self, config: ConfigParser) -> dict[str, Any]:
                 if key and config.has_section(f"formatter_{key}"):
                     result["formatters"][key] = self._parse_formatter_section(config, f"formatter_{key}")
 
-        # Parse handlers  
+        # Parse handlers
         if config.has_section("handlers"):
             handler_keys = config.get("handlers", "keys", fallback="").split(",")
             for key in handler_keys:
@@ -583,7 +597,7 @@ def _convert_ini_to_dictconfig(self, config: ConfigParser) -> dict[str, Any]:
                         result["root"] = self._parse_logger_section(config, f"logger_{key}", is_root=True)
                     else:
                         result["loggers"][key] = self._parse_logger_section(config, f"logger_{key}", is_root=False)
-        
+
         # Parse global settings if they exist
         if config.has_option("DEFAULT", "disable_existing_loggers"):
             result["disable_existing_loggers"] = config.getboolean("DEFAULT", "disable_existing_loggers")
@@ -593,7 +607,7 @@ def _convert_ini_to_dictconfig(self, config: ConfigParser) -> dict[str, Any]:
     def _parse_formatter_section(self, config: ConfigParser, section: str) -> dict[str, Any]:
         """Parse a formatter section."""
         formatter = {}
-        
+
         if config.has_option(section, "format"):
             formatter["format"] = config.get(section, "format")
         if config.has_option(section, "datefmt"):
@@ -606,7 +620,24 @@ def _parse_formatter_section(self, config: ConfigParser, section: str) -> dict[s
             formatter["validate"] = config.getboolean(section, "validate")
         if config.has_option(section, "class"):
             formatter["class"] = config.get(section, "class")
-        
+
+        # Add any other options as extra fields
+        for option in config.options(section):
+            if option not in ["format", "datefmt", "style", "validate", "class"]:
+                value = config.get(section, option)
+                # Try to convert to appropriate type
+                try:
+                    # Try boolean
+                    if value.lower() in ["true", "false"]:
+                        formatter[option] = config.getboolean(section, option)
+                    # Try integer
+                    elif value.isdigit():
+                        formatter[option] = config.getint(section, option)
+                    else:
+                        formatter[option] = value
+                except (ValueError, AttributeError):
+                    formatter[option] = value
+
         return formatter
 
     def _parse_handler_section(self, config: ConfigParser, section: str) -> dict[str, Any]:
@@ -639,15 +670,15 @@ def _parse_handler_section(self, config: ConfigParser, section: str) -> dict[str
             # Handle args - this is complex as it can be a Python expression
             args_str = config.get(section, "args")
             # For basic cases, try to parse as a simple tuple
-            if args_str == "()" or args_str == "()":
+            if args_str == "()":
                 handler["args"] = []
             elif args_str.startswith("(") and args_str.endswith(")"):
                 # Store as string - the logging module will evaluate it
                 handler["args"] = args_str
-        
+
         # Add any other options as extra fields
         for option in config.options(section):
-            if option not in ["class", "level", "formatter", "stream", "filename", "mode", 
+            if option not in ["class", "level", "formatter", "stream", "filename", "mode",
                             "maxBytes", "backupCount", "when", "interval", "utc", "args"]:
                 value = config.get(section, option)
                 # Try to convert to appropriate type
@@ -704,7 +735,7 @@ class LoggingSettings(BaseSettings):
     4. logging.toml - TOML configuration file  
     5. logging.ini - INI configuration file (logging.config.fileConfig format)
     6. pyproject.toml [tool.logging] section (lowest priority)
-    
+
     The model_dump() method returns a dictionary that can be passed
     directly to logging.config.dictConfig().
     """
@@ -713,15 +744,15 @@ class LoggingSettings(BaseSettings):
         default=1,
         description="Configuration schema version"
     )
-    formatters: dict[str, FormatterConfig] = Field(
+    formatters: dict[str, FormatterConfig | CallableFactoryConfig] = Field(
         default_factory=dict,
         description="Formatter configurations"
     )
-    filters: dict[str, FilterConfig] = Field(
+    filters: dict[str, FilterConfig | CallableFactoryConfig] = Field(
         default_factory=dict,
         description="Filter configurations"
     )
-    handlers: dict[str, Any] = Field(
+    handlers: dict[str, HandlerConfig | CallableFactoryConfig] = Field(
         default_factory=dict,
         description="Handler configurations"
     )
@@ -804,7 +835,7 @@ def settings_customise_sources(
                         sources.append(JsonConfigSettingsSource(settings_cls, json_file=json_file))
                     elif Path("logging.json").exists():
                         sources.append(JsonConfigSettingsSource(settings_cls, json_file="logging.json"))
-                    
+
                     if toml_file and Path(toml_file).exists():
                         sources.append(TomlConfigSettingsSource(settings_cls, toml_file=toml_file))
                     elif Path("logging.toml").exists():
@@ -814,7 +845,7 @@ def settings_customise_sources(
                         sources.append(IniConfigSettingsSource(settings_cls, ini_file=ini_file))
                     elif Path("logging.ini").exists():
                         sources.append(IniConfigSettingsSource(settings_cls, ini_file="logging.ini"))
-                    
+
                     # Add pyproject.toml source (lowest priority file)
                     if Path("pyproject.toml").exists():
                         sources.append(
@@ -938,7 +969,7 @@ def model_dump(self, **kwargs) -> dict[str, Any]:
                 else:
                     handlers[name] = handler
             data["handlers"] = handlers
-        
+
         # Convert formatter models to dicts
         if "formatters" in data:
             formatters = {}
@@ -958,7 +989,7 @@ def model_dump(self, **kwargs) -> dict[str, Any]:
                 else:
                     filters[name] = filter_config
             data["filters"] = filters
-        
+
         # Convert logger models to dicts
         if "loggers" in data:
             loggers = {}
@@ -968,7 +999,7 @@ def model_dump(self, **kwargs) -> dict[str, Any]:
                 else:
                     loggers[name] = logger
             data["loggers"] = loggers
-        
+
         # Convert root logger to dict
         if "root" in data and isinstance(data["root"], BaseModel):
             data["root"] = data["root"].model_dump(exclude_none=True)

tests/test_complex_scenarios.py

@@ -76,3 +76,41 @@ def test_filters_configuration():
     assert config_dict["filters"]["myfilter"]["name"] == "myapp.specific"
     assert config_dict["filters"]["custom"]["class"] == "myapp.filters.CustomFilter"
     assert config_dict["handlers"]["filtered"]["filters"] == ["myfilter", "custom"]
+
+
+def test_callable_configuration():
+    """Test configuration with filters."""
+    settings = LoggingSettings(
+        filters={
+            "myfilter": {
+                "()": "myapp.filter",
+                "arg1": "val1"
+            },
+        },
+        formatters={
+            "myformatter": {
+                "()": "myapp.formatter",
+                "arg2": "val2"
+            },
+        },
+        handlers={
+            "myhandler": {
+                "()": "myapp.handler",
+                "arg3": "val3"
+            }
+        }
+    )
+    
+    config_dict = settings.model_dump()
+
+    assert "myfilter" in config_dict["filters"]
+    assert config_dict["filters"]["myfilter"]["()"] == "myapp.filter"
+    assert config_dict["filters"]["myfilter"]["arg1"] == "val1"
+
+    assert "myformatter" in config_dict["formatters"]
+    assert config_dict["formatters"]["myformatter"]["()"] == "myapp.formatter"
+    assert config_dict["formatters"]["myformatter"]["arg2"] == "val2"
+
+    assert "myhandler" in config_dict["handlers"]
+    assert config_dict["handlers"]["myhandler"]["()"] == "myapp.handler"
+    assert config_dict["handlers"]["myhandler"]["arg3"] == "val3"

tests/test_direct_instantiation.py

@@ -1,7 +1,6 @@
 """Test direct instantiation of models with class_ field."""
 
 import pytest
-from pydantic import ValidationError
 
 from pydantic_settings_logging import (
     BaseHandlerConfig,
@@ -134,7 +133,7 @@ def test_mixed_instantiation_styles():
         "delay": False
     }
     assert data == expected
-    
+
     # Recreate from dict (should use aliases)
     handler2 = FileHandlerConfig.model_validate(data)
     assert handler2.class_ == "logging.FileHandler"

tests/test_ini_compatibility.py

@@ -7,7 +7,7 @@
 def test_complex_ini_configuration(tmp_logging_dir, multilogger_ini_config, config_file_factory):
     """Test loading complex INI configuration with multiple loggers."""
     config_file_factory("logging.ini", multilogger_ini_config, "ini")
-    
+
     with change_directory(tmp_logging_dir):
         settings = LoggingSettings()
         config_dict = settings.model_dump()
@@ -41,10 +41,10 @@ def test_ini_edge_cases(tmp_logging_dir, config_file_factory):
 keys=root
 
 [handlers]
-keys=rotatingHandler
+keys=rotatingHandler,customHandler
 
 [formatters]
-keys=detailedFormatter
+keys=detailedFormatter,customFormatter
 
 [logger_root]
 level=DEBUG
@@ -60,10 +60,18 @@ def test_ini_edge_cases(tmp_logging_dir, config_file_factory):
 mode=a
 encoding=utf-8
 
+[handler_customHandler]
+()=myapp.handler_factory
+arg1=val1
+
 [formatter_detailedFormatter]
 format=%(asctime)s [%(process)d] %(name)s %(levelname)s: %(message)s
 datefmt=%Y-%m-%d %H:%M:%S
 validate=true
+
+[formatter_customFormatter]
+()=myapp.formatter_factory
+arg2=val2
 """
     config_file_factory("logging.ini", ini_content, "ini")
 
@@ -78,9 +86,17 @@ def test_ini_edge_cases(tmp_logging_dir, config_file_factory):
         assert handler["backupCount"] == 3
         assert handler["mode"] == "a"
         assert handler["encoding"] == "utf-8"
-        
+
+        handler = config_dict["handlers"]["customHandler"]
+        assert handler["()"] == "myapp.handler_factory"
+        assert handler["arg1"] == "val1"
+
         # Check formatter with validation
         formatter = config_dict["formatters"]["detailedFormatter"]
         assert formatter["format"] == "%(asctime)s [%(process)d] %(name)s %(levelname)s: %(message)s"
         assert formatter["datefmt"] == "%Y-%m-%d %H:%M:%S"
         assert formatter["validate"] is True
+
+        formatter = config_dict["formatters"]["customFormatter"]
+        assert formatter["()"] == "myapp.formatter_factory"
+        assert formatter["arg2"] == "val2"

tests/test_priority_order.py

@@ -1,7 +1,5 @@
 """Tests for configuration source priority order."""
 
-import json
-
 from conftest import change_directory
 from pydantic_settings_logging import LoggingSettings