From 5cc00895ad75f65c9557c08cd20cbdd83dcb86cf Mon Sep 17 00:00:00 2001 From: Federico Bello Date: Wed, 26 Nov 2025 15:28:06 -0300 Subject: [PATCH 1/4] feat: implement lazy loading in base settings --- pydantic_settings/main.py | 123 ++++++++++++++++++++++++++++++ pydantic_settings/sources/base.py | 13 +++- pydantic_settings/sources/lazy.py | 79 +++++++++++++++++++ 3 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 pydantic_settings/sources/lazy.py diff --git a/pydantic_settings/main.py b/pydantic_settings/main.py index 29a09997..ee0f1939 100644 --- a/pydantic_settings/main.py +++ b/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 diff --git a/pydantic_settings/sources/base.py b/pydantic_settings/sources/base.py index 4e733f0f..0f7dcdf6 100644 --- a/pydantic_settings/sources/base.py +++ b/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,7 +552,6 @@ 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 @@ -549,6 +559,7 @@ def __call__(self) -> dict[str, Any]: 'ConfigFileSourceMixin', 'DefaultSettingsSource', 'InitSettingsSource', + 'LazyMapping', 'PydanticBaseEnvSettingsSource', 'PydanticBaseSettingsSource', 'SettingsError', diff --git a/pydantic_settings/sources/lazy.py b/pydantic_settings/sources/lazy.py new file mode 100644 index 00000000..8f304b3d --- /dev/null +++ b/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'] From 52ee47f873dd755fc72c018f4ba6c7b8f370bd88 Mon Sep 17 00:00:00 2001 From: Federico Bello Date: Thu, 27 Nov 2025 11:02:29 -0300 Subject: [PATCH 2/4] feat: add lazy loading in gcp --- pydantic_settings/sources/providers/gcp.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pydantic_settings/sources/providers/gcp.py b/pydantic_settings/sources/providers/gcp.py index 859919b5..c63eed80 100644 --- a/pydantic_settings/sources/providers/gcp.py +++ b/pydantic_settings/sources/providers/gcp.py @@ -102,6 +102,7 @@ def __init__( env_parse_enums: bool | None = None, secret_client: SecretManagerServiceClient | None = None, case_sensitive: bool | None = True, + lazy_load: bool | None = None, ) -> None: # Import Google Packages if they haven't already been imported if SecretManagerServiceClient is None or Credentials is None or google_auth_default is None: @@ -131,7 +132,6 @@ def __init__( self._secret_client = secret_client else: self._secret_client = SecretManagerServiceClient(credentials=self._credentials) - super().__init__( settings_cls, case_sensitive=case_sensitive, @@ -140,6 +140,8 @@ def __init__( env_parse_none_str=env_parse_none_str, env_parse_enums=env_parse_enums, ) + # Set lazy_load after initialization since GCP-specific feature + self.lazy_load = lazy_load if lazy_load is not None else self.config.get('lazy_load', False) def _load_env_vars(self) -> Mapping[str, str | None]: return GoogleSecretManagerMapping( From adaa6c2abe9723bb8e4aa45a1c860073eac0eae9 Mon Sep 17 00:00:00 2001 From: Federico Bello Date: Thu, 27 Nov 2025 12:41:35 -0300 Subject: [PATCH 3/4] doc: add lazy loading in GCP Secret Manager --- docs/index.md | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/docs/index.md b/docs/index.md index f7073848..e46789a7 100644 --- a/docs/index.md +++ b/docs/index.md @@ -2392,6 +2392,74 @@ For nested models, Secret Manager supports the `env_nested_delimiter` setting as For more details on creating and managing secrets in Google Cloud Secret Manager, see the [official Google Cloud documentation](https://cloud.google.com/secret-manager/docs). +### Lazy Loading + +Lazy loading defers field value resolution until fields are actually accessed, rather than eagerly fetching all values during settings initialization. This is particularly useful when working with Google Cloud Secret Manager where each field access triggers an API call, avoiding unnecessary network requests for fields that may never be used. + + +#### Basic Usage + +You can enable lazy loading for Google Cloud Secret Manager via the `lazy_load` parameter when configuring `GoogleSecretManagerSettingsSource`: + +```py +import os + +from pydantic_settings import ( + BaseSettings, + GoogleSecretManagerSettingsSource, + PydanticBaseSettingsSource, +) + + +class Settings(BaseSettings): + secret1: str = '' + secret2: str = '' + + @classmethod + def settings_customise_sources( + cls, + settings_cls: type[BaseSettings], + init_settings: PydanticBaseSettingsSource, + env_settings: PydanticBaseSettingsSource, + dotenv_settings: PydanticBaseSettingsSource, + file_secret_settings: PydanticBaseSettingsSource, + ) -> tuple[PydanticBaseSettingsSource, ...]: + gcp_settings = GoogleSecretManagerSettingsSource( + settings_cls, + project_id=os.environ.get('GCP_PROJECT_ID', 'my-project'), + lazy_load=True, + ) + return ( + init_settings, + env_settings, + dotenv_settings, + gcp_settings, + file_secret_settings, + ) +``` + +When initializing `Settings()` the secrets will not be fetched. When accessing a secret for the first time, for example `secret1`, an API call will be made to fetch that secret, and will then be catched. Next access to that same secret will not trigger an API call, but accessing another one will. Operations that require all secrets, like `model_dump` triggers the fetching of all secrets. + +#### Behavior and Caching + +When lazy loading is enabled: + +1. **Initialization**: Settings are created with minimal overhead. Sources return empty dictionaries instead of eagerly fetching all values. + +2. **First Access**: When you access a field for the first time (e.g., `settings.api_key`), the value is fetched from the configured source and cached in memory. + +3. **Subsequent Access**: Accessing the same field again returns the cached value without making another API call. + +4. **All Fields**: Iteration over all fields (via `model_dump()`, etc.) will trigger resolution of all fields at once. + +***When to use lazy loading:** + +* Your settings have many fields but your application only uses a subset of them +* You want to reduce initialization time and API call costs +* Network latency to GCP Secret Manager is significant + + + ## Other settings source Other settings sources are available for common configuration files: From 3894a89c878ce034586c796f04d987ad7f6cba3d Mon Sep 17 00:00:00 2001 From: Federico Bello Date: Thu, 27 Nov 2025 12:42:43 -0300 Subject: [PATCH 4/4] test: add tests for lazy mapping and lazy load in GCP --- tests/test_lazy_mapping.py | 446 ++++++++++++++++++++++++ tests/test_settings.py | 327 +++++++++++++++++ tests/test_source_gcp_secret_manager.py | 83 +++++ 3 files changed, 856 insertions(+) create mode 100644 tests/test_lazy_mapping.py diff --git a/tests/test_lazy_mapping.py b/tests/test_lazy_mapping.py new file mode 100644 index 00000000..96df87e2 --- /dev/null +++ b/tests/test_lazy_mapping.py @@ -0,0 +1,446 @@ +""" +Test pydantic_settings lazy loading functionality. + +Lazy loading defers field value resolution until the fields are accessed, +rather than eagerly evaluating all fields during settings initialization. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from pydantic_settings import ( + BaseSettings, + PydanticBaseSettingsSource, +) +from pydantic_settings.sources.lazy import LazyMapping + + +class TestLazyMapping: + """Test LazyMapping class behavior.""" + + def test_lazy_mapping_init(self): + """Test LazyMapping initialization.""" + source = MagicMock(spec=PydanticBaseSettingsSource) + mapping = LazyMapping(source) + assert mapping._source is source + assert mapping._cached_values == {} + + def test_lazy_mapping_getitem_with_caching(self): + """Test LazyMapping caches values after first access.""" + source = MagicMock() + + # Mock settings class with a field + settings_cls = MagicMock() + field_info = MagicMock() + settings_cls.model_fields = {'test_field': field_info} + source.settings_cls = settings_cls + + # Mock the field resolution + source._get_resolved_field_value.return_value = ('test-value', None, False) + source.prepare_field_value.return_value = 'prepared-value' + + mapping = LazyMapping(source) + + # First access should call the resolution methods + value1 = mapping['test_field'] + assert value1 == 'prepared-value' + assert source._get_resolved_field_value.call_count == 1 + + # Second access should use cached value + value2 = mapping['test_field'] + assert value2 == 'prepared-value' + assert source._get_resolved_field_value.call_count == 1 # Not called again + + def test_lazy_mapping_getitem_key_not_found(self): + """Test LazyMapping raises KeyError for missing keys.""" + source = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + + with pytest.raises(KeyError): + _ = mapping['nonexistent'] + + def test_lazy_mapping_iter(self): + """Test LazyMapping iteration returns all field names.""" + source = MagicMock() + + # Mock settings class with multiple fields + field_info1 = MagicMock() + field_info2 = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info1, 'field2': field_info2} + source.settings_cls = settings_cls + + # Mock alias names function + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.side_effect = [(['alias1'], None), (['alias2'], None)] + + mapping = LazyMapping(source) + keys = list(mapping) + + assert 'alias1' in keys + assert 'field1' in keys or 'alias2' in keys + assert 'field2' in keys or 'alias2' in keys + + def test_lazy_mapping_len(self): + """Test LazyMapping returns correct length.""" + source = MagicMock(spec=PydanticBaseSettingsSource) + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': MagicMock(), 'field2': MagicMock(), 'field3': MagicMock()} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + assert len(mapping) == 3 + + def test_lazy_mapping_copy(self): + """Test LazyMapping.copy() preserves cached values.""" + source = MagicMock(spec=PydanticBaseSettingsSource) + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': MagicMock()} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + mapping._cached_values['field1'] = 'cached-value' + + copied = mapping.copy() + + assert isinstance(copied, LazyMapping) + assert copied._source is source + assert copied._cached_values == {'field1': 'cached-value'} + # Ensure it's a shallow copy of the dict + assert copied._cached_values is not mapping._cached_values + + def test_lazy_mapping_items(self): + """Test LazyMapping.items() yields accessible key-value pairs.""" + source = MagicMock() + + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'test_field': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('value', None, False) + source.prepare_field_value.return_value = 'prepared-value' + + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.return_value = (['test_field'], None) + + mapping = LazyMapping(source) + items = dict(mapping.items()) + + assert 'test_field' in items + assert items['test_field'] == 'prepared-value' + + +class TestLazyLoadingWithDeepUpdate: + """Test LazyMapping behavior with deep_update to ensure lazy behavior is preserved.""" + + def test_lazy_mapping_deep_update_preserves_lazy_behavior(self): + """Test that deep_update with LazyMapping copy() preserves lazy behavior.""" + from pydantic._internal._utils import deep_update + + source = MagicMock() + settings_cls = MagicMock() + field_info = MagicMock() + settings_cls.model_fields = {'field1': field_info, 'field2': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('value', None, False) + source.prepare_field_value.side_effect = lambda fname, fi, v, _: f'prepared-{fname}' + + # Create two LazyMappings + mapping2 = LazyMapping(source) + + # Simulate deep_update behavior + result = deep_update({'field1': 'eager-value'}, mapping2.copy()) + + # The result should contain the lazily evaluated value from mapping2 + # and the copy() should have returned a LazyMapping, not a regular dict + assert isinstance(result.get('field1'), str) + + def test_lazy_mapping_copy_maintains_source_reference(self): + """Test that LazyMapping.copy() maintains reference to the same source.""" + source = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + copied = mapping.copy() + + # Both should reference the same source object + assert mapping._source is copied._source + + +class TestLazyLoadingWithConfigFiles: + """Test lazy loading compatibility with config file sources.""" + + def test_lazy_mapping_with_dict_like_interface(self): + """Test that LazyMapping implements proper dict-like interface.""" + source = MagicMock() + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('test-value', None, False) + source.prepare_field_value.return_value = 'prepared-value' + + mapping = LazyMapping(source) + + # Test Mapping interface methods + assert 'field1' in list(mapping) + assert len(mapping) == 1 + assert mapping['field1'] == 'prepared-value' + + def test_lazy_mapping_error_on_missing_field(self): + """Test LazyMapping raises SettingsError for missing fields during resolution.""" + from pydantic_settings.exceptions import SettingsError + + source = MagicMock() + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info} + source.settings_cls = settings_cls + source.__class__.__name__ = 'MockSource' + + # Simulate an error during field value resolution + source._get_resolved_field_value.side_effect = ValueError('Connection failed') + + mapping = LazyMapping(source) + + # Accessing the field should raise SettingsError (not ValueError) + with pytest.raises(SettingsError, match='error getting value for field "field1"'): + _ = mapping['field1'] + + +class TestLazyLoadingSourcesWithParameter: + """Test that GCP sources accept lazy_load parameter.""" + + def test_gcp_secrets_source_accepts_lazy_load(self): + """Test GoogleSecretManagerSettingsSource accepts lazy_load parameter.""" + from pydantic_settings.sources.providers.gcp import GoogleSecretManagerSettingsSource + + class TestSettings(BaseSettings): + field: str = 'default' + + try: + # This will fail if google-cloud-secret-manager is not installed + import inspect + + sig = inspect.signature(GoogleSecretManagerSettingsSource.__init__) + assert 'lazy_load' in sig.parameters + except ImportError: + # gcp not installed, skip + pytest.skip('google-cloud-secret-manager not installed') + + +class TestGCPSecretManagerLazyLoading: + """Unit tests for lazy_load behavior in GoogleSecretManagerSettingsSource.""" + + def test_returns_lazy_mapping_when_lazy_load_true(self): + """Test GoogleSecretManagerSettingsSource stores LazyMapping when lazy_load=True.""" + try: + from pydantic_settings.sources.providers.gcp import GoogleSecretManagerSettingsSource + except ImportError: + pytest.skip('google-cloud-secret-manager not installed') + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=True) + result = source() + assert isinstance(result, dict) + assert len(result) == 0 + assert hasattr(source, '_lazy_mapping') + assert isinstance(source._lazy_mapping, LazyMapping) + + def test_returns_dict_when_lazy_load_false(self): + """Test GoogleSecretManagerSettingsSource returns dict when lazy_load=False.""" + try: + from pydantic_settings.sources.providers.gcp import GoogleSecretManagerSettingsSource + except ImportError: + pytest.skip('google-cloud-secret-manager not installed') + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=False) + result = source() + assert isinstance(result, dict) + assert not hasattr(source, '_lazy_mapping') or source._lazy_mapping is None + + def test_lazy_mapping_defers_resolution(self): + """Test GoogleSecretManagerSettingsSource LazyMapping defers resolution.""" + try: + from pydantic_settings.sources.providers.gcp import GoogleSecretManagerSettingsSource + except ImportError: + pytest.skip('google-cloud-secret-manager not installed') + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=True) + + with patch.object( + source, '_get_resolved_field_value', wraps=source._get_resolved_field_value + ) as mock_resolve: + source() + assert mock_resolve.call_count == 0 + + lazy_mapping = source._lazy_mapping + try: + _ = lazy_mapping['field'] + assert mock_resolve.call_count > 0 + except KeyError: + assert mock_resolve.call_count > 0 + + +class TestLazyMappingAdditionalMethods: + """Test additional LazyMapping methods and edge cases.""" + + def test_lazy_mapping_get_method(self): + """Test LazyMapping.get() method.""" + source = MagicMock() + settings_cls = MagicMock() + field_info = MagicMock() + settings_cls.model_fields = {'field1': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('value1', None, False) + source.prepare_field_value.return_value = 'prepared-value1' + + mapping = LazyMapping(source) + + # Test get with existing key + assert mapping.get('field1') == 'prepared-value1' + + # Test get with non-existing key and default + assert mapping.get('nonexistent', 'default') == 'default' + + # Test get with non-existing key and no default + assert mapping.get('nonexistent') is None + + def test_lazy_mapping_keys_method(self): + """Test LazyMapping.keys() method.""" + source = MagicMock() + field_info1 = MagicMock() + field_info2 = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info1, 'field2': field_info2} + source.settings_cls = settings_cls + + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.side_effect = [(['field1'], None), (['field2'], None)] + + mapping = LazyMapping(source) + keys = list(mapping.keys()) + + assert 'field1' in keys + assert 'field2' in keys + assert len(keys) == 2 + + def test_lazy_mapping_values_method(self): + """Test LazyMapping.values() method.""" + source = MagicMock() + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('value1', None, False) + source.prepare_field_value.return_value = 'prepared-value1' + + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.return_value = (['field1'], None) + + mapping = LazyMapping(source) + values = list(mapping.values()) + + assert 'prepared-value1' in values + assert len(values) == 1 + + def test_lazy_mapping_contains_method(self): + """Test LazyMapping __contains__ method.""" + source = MagicMock() + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('value1', None, False) + source.prepare_field_value.return_value = 'prepared-value1' + + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.return_value = (['field1'], None) + + mapping = LazyMapping(source) + + assert 'field1' in mapping + assert 'nonexistent' not in mapping + + def test_lazy_mapping_bool_conversion(self): + """Test LazyMapping __bool__ method.""" + source = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field1': MagicMock()} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + + # Non-empty mapping should be truthy + assert bool(mapping) is True + + # Empty mapping should be falsy + settings_cls.model_fields = {} + mapping_empty = LazyMapping(source) + assert bool(mapping_empty) is False + + def test_lazy_mapping_with_alias_resolution(self): + """Test LazyMapping resolves fields by alias names.""" + source = MagicMock() + field_info = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {'field_name': field_info} + source.settings_cls = settings_cls + + source._get_resolved_field_value.return_value = ('alias_value', None, False) + source.prepare_field_value.return_value = 'prepared-alias-value' + + with patch('pydantic_settings.sources.lazy._get_alias_names') as mock_alias: + mock_alias.return_value = (['field_alias'], None) + + mapping = LazyMapping(source) + + # Access by alias should resolve correctly + assert mapping['field_alias'] == 'prepared-alias-value' + + def test_lazy_mapping_field_not_found_raises_key_error(self): + """Test LazyMapping raises KeyError for completely missing fields.""" + source = MagicMock() + settings_cls = MagicMock() + settings_cls.model_fields = {} + source.settings_cls = settings_cls + + mapping = LazyMapping(source) + + with pytest.raises(KeyError): + _ = mapping['nonexistent_field'] diff --git a/tests/test_settings.py b/tests/test_settings.py index f7c84059..68f09abf 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -3264,3 +3264,330 @@ class StrictSettings(BaseSettings, env_nested_delimiter='__', strict=True): 'my_int': 1, }, } + + +def test_lazy_loading_with_field_default_factory(): + """Test lazy loading when field has a default_factory and value is None.""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + # Return None so field ends up as None, triggering default_factory check + return {'field_with_factory': None} + + @property + def _lazy_mapping(self): + return {'field_with_factory': 'lazy_value'} + + class Settings(BaseSettings): + field_with_factory: str | None = Field(default=None) + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # Access the field to trigger lazy loading + value = settings.field_with_factory + assert value == 'lazy_value' + + +def test_lazy_loading_accessing_nonexistent_field_in_lazy_sources(): + """Test lazy loading when accessing a field not in lazy sources (KeyError path).""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + return {} + + @property + def _lazy_mapping(self): + return {} + + class Settings(BaseSettings): + field1: str = 'default' + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # Accessing the field should not raise, just return default + value = settings.field1 + assert value == 'default' + + +def test_lazy_loading_without_lazy_sources_attribute(): + """Test __getattribute__ when _lazy_sources is not set (AttributeError path).""" + + class Settings(BaseSettings): + field1: str = 'default' + + settings = Settings() + # Should work fine even without _lazy_sources + value = settings.field1 + assert value == 'default' + + +def test_model_dump_with_lazy_sources_and_defaults(): + """Test model_dump includes lazy-loaded values for fields at default.""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + return {} + + @property + def _lazy_mapping(self): + return {'field1': 'lazy_value1', 'field2': 'lazy_value2'} + + class Settings(BaseSettings): + field1: str = 'default1' + field2: str = 'default2' + field3: str = 'default3' + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + dump = settings.model_dump() + # Fields in lazy mapping should be loaded + assert dump['field1'] == 'lazy_value1' + assert dump['field2'] == 'lazy_value2' + # Field not in lazy mapping should keep default + assert dump['field3'] == 'default3' + + +def test_model_dump_without_lazy_sources(): + """Test model_dump when _lazy_sources is not set.""" + + class Settings(BaseSettings): + field1: str = 'default1' + + settings = Settings() + dump = settings.model_dump() + assert dump['field1'] == 'default1' + + +def test_lazy_sources_with_required_field(): + """Test lazy loading with required fields (no default).""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + # Provide the required field via the source + return {'required_field': 'source_value'} + + @property + def _lazy_mapping(self): + return {'required_field': 'lazy_value'} + + class Settings(BaseSettings): + required_field: str + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # The required field gets source_value from __call__ + assert settings.required_field == 'source_value' + + +def test_model_dump_with_required_field_lazy_sources(): + """Test model_dump with required fields and lazy sources.""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + # Provide the required field + return {'required_field': 'source_value'} + + @property + def _lazy_mapping(self): + # Also add lazy value for lazy loading + return {'required_field': 'lazy_required_value'} + + class Settings(BaseSettings): + required_field: str + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # The required field gets source_value from __call__ (higher priority) + assert settings.required_field == 'source_value' + + +def test_empty_sources_returns_empty_dict(): + """Test _settings_build_values returns empty dict when sources is empty.""" + + class Settings(BaseSettings): + field1: str = 'default' + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + # Return empty tuple of sources + return () + + # This should work and return empty dict from _settings_build_values + settings = Settings() + # Should use default since no sources provide values + assert settings.field1 == 'default' + + +def test_lazy_sources_collected_from_source(): + """Test that lazy mappings are collected from sources with _lazy_mapping attribute.""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + return {'regular_field': 'regular_value'} + + @property + def _lazy_mapping(self): + return {'lazy_field': 'lazy_value'} + + class Settings(BaseSettings): + regular_field: str = 'default' + lazy_field: str = 'default' + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # Regular field should get value from source return value + assert settings.regular_field == 'regular_value' + # Lazy field should get value from lazy mapping + assert settings.lazy_field == 'lazy_value' + + +def test_cli_app_async_command_exception_handling(): + """Test CliApp._run_cli_cmd handles async exceptions properly.""" + from pydantic_settings import CliApp + + class TestModel(BaseSettings): + name: str = 'test' + + async def cli_cmd(self): + # Simulate async work + pass + + # This should work without raising - pass empty list to avoid pytest args + model = CliApp.run(TestModel, cli_args=[], cli_exit_on_error=False) + assert model.name == 'test' + + +def test_async_command_sync_version(): + """Test CliApp._run_cli_cmd with synchronous command.""" + from pydantic_settings import CliApp + + class TestModel(BaseSettings): + name: str = 'test' + + def cli_cmd(self): + pass # Synchronous command + + # This should work without issues - pass empty list to avoid pytest args + model = CliApp.run(TestModel, cli_args=[], cli_exit_on_error=False) + assert model.name == 'test' + + +def test_lazy_loading_field_with_none_default(): + """Test lazy loading with field that has None as default.""" + from pydantic_settings.sources import PydanticBaseSettingsSource + + class MockLazySource(PydanticBaseSettingsSource): + def get_field_value(self, field, field_name): + return None, None, False + + def __call__(self) -> dict[str, Any]: + return {} + + @property + def _lazy_mapping(self): + return {'optional_field': 'lazy_value'} + + class Settings(BaseSettings): + optional_field: str | None = None + + @classmethod + def settings_customise_sources( + cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings + ): + return (MockLazySource(settings_cls),) + ( + init_settings, + env_settings, + dotenv_settings, + file_secret_settings, + ) + + settings = Settings() + # Should get lazy value since default is None + value = settings.optional_field + assert value == 'lazy_value' diff --git a/tests/test_source_gcp_secret_manager.py b/tests/test_source_gcp_secret_manager.py index 1ed76468..cc05ba08 100644 --- a/tests/test_source_gcp_secret_manager.py +++ b/tests/test_source_gcp_secret_manager.py @@ -2,12 +2,15 @@ Test pydantic_settings.GoogleSecretSettingsSource """ +from unittest.mock import MagicMock, patch + import pytest from pydantic import Field from pytest_mock import MockerFixture from pydantic_settings import BaseSettings, PydanticBaseSettingsSource from pydantic_settings.sources import GoogleSecretManagerSettingsSource +from pydantic_settings.sources.lazy import LazyMapping from pydantic_settings.sources.providers.gcp import GoogleSecretManagerMapping, import_gcp_secret_manager try: @@ -242,3 +245,83 @@ def test_secret_manager_mapping_list_secrets_error(self, secret_manager_mapping, with pytest.raises(Exception, match='Permission denied'): _ = secret_manager_mapping._secret_names + + +@pytest.mark.skipif(not gcp_secret_manager, reason='pydantic-settings[gcp-secret-manager] is not installed') +class TestGCPSecretManagerLazyLoading: + """Unit tests for lazy_load behavior in GoogleSecretManagerSettingsSource.""" + + def test_returns_lazy_mapping_when_lazy_load_true(self): + """Test GoogleSecretManagerSettingsSource stores LazyMapping when lazy_load=True.""" + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=True) + result = source() + assert isinstance(result, dict) + assert len(result) == 0 + assert hasattr(source, '_lazy_mapping') + assert isinstance(source._lazy_mapping, LazyMapping) + + def test_returns_dict_when_lazy_load_false(self): + """Test GoogleSecretManagerSettingsSource returns dict when lazy_load=False.""" + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=False) + result = source() + assert isinstance(result, dict) + assert not hasattr(source, '_lazy_mapping') or source._lazy_mapping is None + + def test_lazy_mapping_defers_resolution(self): + """Test GoogleSecretManagerSettingsSource LazyMapping defers resolution.""" + + class TestSettings(BaseSettings): + field: str = 'default' + + with patch( + 'pydantic_settings.sources.providers.gcp.google_auth_default', return_value=(MagicMock(), 'test-project') + ): + with patch('pydantic_settings.sources.providers.gcp.SecretManagerServiceClient'): + source = GoogleSecretManagerSettingsSource(TestSettings, lazy_load=True) + + with patch.object( + source, '_get_resolved_field_value', wraps=source._get_resolved_field_value + ) as mock_resolve: + source() + assert mock_resolve.call_count == 0 + + lazy_mapping = source._lazy_mapping + try: + _ = lazy_mapping['field'] + assert mock_resolve.call_count > 0 + except KeyError: + assert mock_resolve.call_count > 0 + + +@pytest.mark.skipif(not gcp_secret_manager, reason='pydantic-settings[gcp-secret-manager] is not installed') +def test_gcp_secrets_source_accepts_lazy_load(): + """Test GoogleSecretManagerSettingsSource accepts lazy_load parameter.""" + + class TestSettings(BaseSettings): + field: str = 'default' + + try: + # This will fail if google-cloud-secret-manager is not installed + import inspect + + sig = inspect.signature(GoogleSecretManagerSettingsSource.__init__) + assert 'lazy_load' in sig.parameters + except ImportError: + # gcp not installed, skip + pytest.skip('google-cloud-secret-manager not installed')