feat: implement lazy loading in base settings

Author: fede-bello

pydantic_settings/main.py

@@ -100,6 +100,7 @@ class SettingsConfigDict(ConfigDict, total=False):
 
     toml_file: PathType | None
     enable_decoding: bool
+    lazy_load: bool
 
 
 # Extend `config_keys` by pydantic settings config keys to
@@ -159,6 +160,8 @@ class BaseSettings(BaseModel):
         _cli_kebab_case: CLI args use kebab case. Defaults to `False`.
         _cli_shortcuts: Mapping of target field name to alias names. Defaults to `None`.
         _secrets_dir: The secret files directory or a sequence of directories. Defaults to `None`.
+        _lazy_load: Defer field value resolution until fields are accessed. When enabled, field values
+            are only fetched from the source when explicitly accessed, not during settings initialization.
     """
 
     def __init__(
@@ -191,6 +194,10 @@ def __init__(
         _secrets_dir: PathType | None = None,
         **values: Any,
     ) -> None:
+        # Temp storage for lazy sources collected during _settings_build_values
+        _temp_lazy_sources: dict[str, Any] = {}
+        __pydantic_self__._temp_lazy_sources = _temp_lazy_sources
+
         super().__init__(
             **__pydantic_self__._settings_build_values(
                 values,
@@ -223,6 +230,117 @@ def __init__(
             )
         )
 
+        # Now that super().__init__() has completed, set the lazy sources on the instance
+        # using object.__setattr__ to bypass any Pydantic restrictions
+        object.__setattr__(__pydantic_self__, '_lazy_sources', _temp_lazy_sources)
+
+    def __getattribute__(self, name: str) -> Any:
+        """Intercept field access to support lazy loading on demand."""
+        # Get the actual value from the model
+        value = super().__getattribute__(name)
+
+        # Return private attributes and methods as-is
+        if name.startswith('_') or callable(value):
+            return value
+
+        # For model fields, try to get value from lazy sources only if not set by
+        # higher-priority sources. We detect this by checking if the value is the field's default.
+        try:
+            model_cls = type(self)
+            if name in model_cls.model_fields:
+                field_info = model_cls.model_fields[name]
+                # Only try lazy sources if the value is the default (wasn't set by higher-priority source)
+                # Check if value is the field's default value
+                is_default = False
+                if field_info.is_required():
+                    # Required fields have no default, so if value is not None, it was set
+                    is_default = value is None
+                elif field_info.default is not None:
+                    is_default = value == field_info.default
+                elif field_info.default_factory is not None:
+                    # For fields with default_factory, comparing to the factory output
+                    # would require calling the factory, so we check if value is unset
+                    is_default = value is None or value == field_info.default
+                else:
+                    is_default = value is None or value == field_info.default
+
+                if is_default:
+                    lazy_sources = object.__getattribute__(self, '_lazy_sources')
+                    for lazy_mapping in lazy_sources.values():
+                        try:
+                            return lazy_mapping[name]
+                        except KeyError:
+                            pass
+        except AttributeError:
+            pass
+
+        return value
+
+    def model_dump(
+        self,
+        *,
+        mode: str | Literal['json', 'python'] = 'python',
+        include: Any = None,
+        exclude: Any = None,
+        context: Any = None,
+        by_alias: bool | None = None,
+        exclude_unset: bool = False,
+        exclude_defaults: bool = False,
+        exclude_none: bool = False,
+        round_trip: bool = False,
+        warnings: Literal['none', 'warn', 'error'] | bool = True,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        """Override model_dump to include cached lazy-loaded values."""
+        # Get base dump from parent class
+        dump = super().model_dump(
+            mode=mode,
+            include=include,
+            exclude=exclude,
+            context=context,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_defaults=exclude_defaults,
+            exclude_none=exclude_none,
+            round_trip=round_trip,
+            warnings=warnings,
+            **kwargs,
+        )
+
+        # Merge lazy values from _lazy_sources, triggering loads if needed
+        try:
+            lazy_sources = object.__getattribute__(self, '_lazy_sources')
+            for source_name, lazy_mapping in lazy_sources.items():
+                # Iterate through all fields in the lazy mapping and load any that are
+                # still at their default value (not set by higher-priority sources)
+                for field_name in lazy_mapping:
+                    # Check if field is still at default in dump
+                    if field_name in type(self).model_fields:
+                        field_info = type(self).model_fields[field_name]
+                        current_value = dump.get(field_name)
+
+                        # Determine if this is still a default value
+                        is_default = False
+                        if field_info.is_required():
+                            is_default = current_value is None
+                        elif field_info.default is not None:
+                            is_default = current_value == field_info.default
+                        else:
+                            is_default = current_value is None or current_value == field_info.default
+
+                        # If still at default, try to load from lazy mapping
+                        if is_default:
+                            try:
+                                dump[field_name] = lazy_mapping[field_name]
+                            except KeyError:
+                                # Field not available in this lazy source, keep default
+                                pass
+        except AttributeError:
+            # _lazy_sources not set (no lazy sources configured) - return base dump
+            pass
+
+        return dump
+
     @classmethod
     def settings_customise_sources(
         cls,
@@ -437,6 +555,11 @@ def _settings_build_values(
                 source_name = source.__name__ if hasattr(source, '__name__') else type(source).__name__
                 source_state = source()
 
+                # Collect lazy mappings from sources for later field access
+                if hasattr(source, '_lazy_mapping'):
+                    temp_lazy_sources = self._temp_lazy_sources
+                    temp_lazy_sources[source_name] = source._lazy_mapping
+
                 if isinstance(source, DefaultSettingsSource):
                     defaults = source_state
 

pydantic_settings/sources/base.py

@@ -20,6 +20,7 @@
 
 from ..exceptions import SettingsError
 from ..utils import _lenient_issubclass
+from .lazy import LazyMapping
 from .types import EnvNoneType, ForceDecode, NoDecode, PathType, PydanticModel, _CliSubCommand
 from .utils import (
     _annotation_is_complex,
@@ -326,6 +327,7 @@ def __init__(
         env_ignore_empty: bool | None = None,
         env_parse_none_str: str | None = None,
         env_parse_enums: bool | None = None,
+        lazy_load: bool | None = None,
     ) -> None:
         super().__init__(settings_cls)
         self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False)
@@ -337,6 +339,7 @@ def __init__(
             env_parse_none_str if env_parse_none_str is not None else self.config.get('env_parse_none_str')
         )
         self.env_parse_enums = env_parse_enums if env_parse_enums is not None else self.config.get('env_parse_enums')
+        self.lazy_load = lazy_load if lazy_load is not None else self.config.get('lazy_load', False)
 
     def _apply_case_sensitive(self, value: str) -> str:
         return value.lower() if not self.case_sensitive else value
@@ -510,6 +513,14 @@ def _get_resolved_field_value(self, field: FieldInfo, field_name: str) -> tuple[
         return field_value, field_key, value_is_complex
 
     def __call__(self) -> dict[str, Any]:
+        # If lazy loading is enabled, defer field resolution to access time
+        if self.lazy_load:
+            # Store the LazyMapping on the source for later retrieval
+            self._lazy_mapping = LazyMapping(self)
+            # Return empty dict to avoid eager evaluation during initialization
+            return {}
+
+        # Otherwise, use eager field loading
         data: dict[str, Any] = {}
 
         for field_name, field in self.settings_cls.model_fields.items():
@@ -541,14 +552,14 @@ def __call__(self) -> dict[str, Any]:
                     data[field_key] = self._replace_field_names_case_insensitively(field, field_value)
                 else:
                     data[field_key] = field_value
-
         return data
 
 
 __all__ = [
     'ConfigFileSourceMixin',
     'DefaultSettingsSource',
     'InitSettingsSource',
+    'LazyMapping',
     'PydanticBaseEnvSettingsSource',
     'PydanticBaseSettingsSource',
     'SettingsError',

pydantic_settings/sources/lazy.py

@@ -0,0 +1,79 @@
+"""Lazy loading support for settings sources."""
+
+from __future__ import annotations as _annotations
+
+from collections.abc import Iterator, Mapping
+from typing import TYPE_CHECKING, Any
+
+from pydantic.fields import FieldInfo
+
+from ..exceptions import SettingsError
+from .utils import _get_alias_names
+
+if TYPE_CHECKING:
+    from .base import PydanticBaseEnvSettingsSource
+
+
+class LazyMapping(Mapping[str, Any]):
+    """Dict-like mapping that defers field value resolution until keys are accessed."""
+
+    def __init__(self, source: PydanticBaseEnvSettingsSource) -> None:
+        """Initialize with a source instance that will compute values on demand."""
+        self._source = source
+        self._cached_values: dict[str, Any] = {}
+
+    def __getitem__(self, key: str) -> Any:
+        """Get a field value, computing it lazily on first access."""
+        # Return cached value if available
+        if key in self._cached_values:
+            return self._cached_values[key]
+
+        # Find the field in the settings class
+        field_name: str | None = None
+        field_info: FieldInfo | None = None
+
+        for fname, finfo in self._source.settings_cls.model_fields.items():
+            alias_names, *_ = _get_alias_names(fname, finfo)
+            if key in alias_names or key == fname:
+                field_name = fname
+                field_info = finfo
+                break
+
+        if field_name is None or field_info is None:
+            raise KeyError(key)
+
+        # Resolve and cache the field value
+        try:
+            field_value, _, value_is_complex = self._source._get_resolved_field_value(field_info, field_name)
+            prepared_value = self._source.prepare_field_value(field_name, field_info, field_value, value_is_complex)
+            self._cached_values[key] = prepared_value
+            return prepared_value
+        except Exception as e:
+            raise SettingsError(
+                f'error getting value for field "{field_name}" from source "{self._source.__class__.__name__}"'
+            ) from e
+
+    def __iter__(self) -> Iterator[str]:
+        """Iterate over all possible field keys."""
+        seen: set[str] = set()
+        for field_name, field_info in self._source.settings_cls.model_fields.items():
+            alias_names, *_ = _get_alias_names(field_name, field_info)
+            for alias in alias_names:
+                if alias not in seen:
+                    seen.add(alias)
+                    yield alias
+            if field_name not in seen:
+                yield field_name
+
+    def __len__(self) -> int:
+        """Return the count of fields in the settings class."""
+        return len(self._source.settings_cls.model_fields)
+
+    def copy(self) -> LazyMapping:
+        """Return a copy to preserve lazy behavior through Pydantic's deep_update()."""
+        new_mapping = LazyMapping(self._source)
+        new_mapping._cached_values = self._cached_values.copy()
+        return new_mapping
+
+
+__all__ = ['LazyMapping']