Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions docs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Comment on lines +2447 to +2448
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two questions:

  1. What happens if other sources' values have more priority than GCP settings source?
  2. What happens if the value provided by a source is not a valid value?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. Not sure in what sense you mean. If I understand your question correctly, higher priority sources shadow lower ones. So if theres a key in a higher priority source, that one will be loaded and the one for GCP won't be consulted
  2. Trying to access will return None. It will enter the get_field_value method in EnvSettingsSource, the field_value will not be found and will return None

PS: left a fix with an issue with the model dump that wasn't loading the lazy fields

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Trying to access will return None. It will enter the get_field_value method in EnvSettingsSource, the field_value will not be found and will return None

I mean if the value is not a valid value for the field. like you defined an int field but the value is string. or you put some limitation on the string length.

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:
Expand Down
123 changes: 123 additions & 0 deletions pydantic_settings/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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__(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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

Expand Down
13 changes: 12 additions & 1 deletion pydantic_settings/sources/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Copy link
Member

@hramezani hramezani Nov 28, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we keep this here? we agreed to enable lazy loading for GCP secret source

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I left it there so eventually is easier to implement for other sources providers and it's not a only-gcp fix. It the parameter is not passed to the base class the logic shouldn't change

) -> None:
super().__init__(settings_cls)
self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False)
Expand All @@ -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
Expand Down Expand Up @@ -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():
Expand Down Expand Up @@ -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',
Expand Down
79 changes: 79 additions & 0 deletions pydantic_settings/sources/lazy.py
Original file line number Diff line number Diff line change
@@ -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']
4 changes: 3 additions & 1 deletion pydantic_settings/sources/providers/gcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand All @@ -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(
Expand Down
Loading