"""Base classes and core functionality for pydantic-settings sources.""" from __future__ import annotations as _annotations import json from abc import ABC, abstractmethod from collections.abc import Callable, Hashable, Mapping, Sequence from dataclasses import asdict, is_dataclass from dataclasses import fields as dataclass_fields from pathlib import Path from typing import TYPE_CHECKING, Any, cast, get_args, get_type_hints from pydantic import AliasChoices, AliasPath, TypeAdapter from pydantic._internal._typing_extra import ( # type: ignore[attr-defined] get_origin, ) from pydantic._internal._utils import deep_update, is_model_class from pydantic.dataclasses import is_pydantic_dataclass from pydantic.fields import FieldInfo from typing_inspection.introspection import is_union_origin from ..exceptions import SettingsError from .types import ( ConfigFileSourceType, EnvNoneType, EnvPrefixTarget, ForceDecode, NoDecode, PydanticModel, _CliSubCommand, ) from .utils import ( InitState, _annotation_is_complex, _get_alias_names, _get_field_annotation_and_metadata, _get_field_metadata, _get_model_config, _get_model_fields, _strip_annotated, _union_is_complex, _validate_by_alias_enabled, _validate_by_name_enabled, _warn_if_field_info_incomplete, ) if TYPE_CHECKING: from pydantic_settings.main import BaseSettings from .types import Traversable def get_subcommand( model: PydanticModel, is_required: bool = True, cli_exit_on_error: bool | None = None, _suppress_errors: list[SettingsError | SystemExit] | None = None, ) -> PydanticModel | None: """ Get the subcommand from a model. Args: model: The model to get the subcommand from. is_required: Determines whether a model must have subcommand set and raises error if not found. Defaults to `True`. cli_exit_on_error: Determines whether this function exits with error if no subcommand is found. Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`. Returns: The subcommand model if found, otherwise `None`. Raises: SystemExit: When no subcommand is found and is_required=`True` and cli_exit_on_error=`True` (the default). SettingsError: When no subcommand is found and is_required=`True` and cli_exit_on_error=`False`. """ model_cls = type(model) if cli_exit_on_error is None and is_model_class(model_cls): model_default = model_cls.model_config.get('cli_exit_on_error') if isinstance(model_default, bool): cli_exit_on_error = model_default if cli_exit_on_error is None: cli_exit_on_error = True subcommands: list[str] = [] for field_name, field_info in _get_model_fields(model_cls).items(): if _CliSubCommand in field_info.metadata: if getattr(model, field_name) is not None: return getattr(model, field_name) subcommands.append(field_name) # TODO: we could warn if no sub command was found and one of `field_info`s # isn't complete. if is_required: error_message = ( f'Error: CLI subcommand is required {{{", ".join(subcommands)}}}' if subcommands else 'Error: CLI subcommand is required but no subcommands were found.' ) err = SystemExit(error_message) if cli_exit_on_error else SettingsError(error_message) if _suppress_errors is None: raise err _suppress_errors.append(err) return None class PydanticBaseSettingsSource(ABC): """ Abstract base class for settings sources, every settings source classes should inherit from it. """ def __init__(self, settings_cls: type[BaseSettings], _init_state: InitState | None = None): self._init_state: InitState = {} if _init_state is None else _init_state self.settings_cls = settings_cls self.config = settings_cls.model_config self._current_state: dict[str, Any] = {} self._settings_sources_data: dict[str, dict[str, Any]] = {} def _set_current_state(self, state: dict[str, Any]) -> None: """ Record the state of settings from the previous settings sources. This should be called right before __call__. """ self._current_state = state def _set_settings_sources_data(self, states: dict[str, dict[str, Any]]) -> None: """ Record the state of settings from all previous settings sources. This should be called right before __call__. """ self._settings_sources_data = states @property def current_state(self) -> dict[str, Any]: """ The current state of the settings, populated by the previous settings sources. """ return self._current_state @property def settings_sources_data(self) -> dict[str, dict[str, Any]]: """ The state of all previous settings sources. """ return self._settings_sources_data @abstractmethod def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: """ Gets the value, the key for model creation, and a flag to determine whether value is complex. This is an abstract method that should be overridden in every settings source classes. Args: field: The field. field_name: The field name. Returns: A tuple that contains the value, key and a flag to determine whether value is complex. """ def field_is_complex(self, field: FieldInfo) -> bool: """ Checks whether a field is complex, in which case it will attempt to be parsed as JSON. Args: field: The field. Returns: Whether the field is complex. """ return _annotation_is_complex(field.annotation, field.metadata, self._init_state) def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any: """ Prepares the value of a field. Args: field_name: The field name. field: The field. value: The value of the field that has to be prepared. value_is_complex: A flag to determine whether value is complex. Returns: The prepared value. """ if value is not None and (self.field_is_complex(field) or value_is_complex): return self.decode_complex_value(field_name, field, value) return value def decode_complex_value(self, field_name: str, field: FieldInfo, value: Any) -> Any: """ Decode the value for a complex field Args: field_name: The field name. field: The field. value: The value of the field that has to be prepared. Returns: The decoded value for further preparation """ metadata = _get_field_metadata(field) if field else [] if field and ( NoDecode in metadata or (self.config.get('enable_decoding') is False and ForceDecode not in metadata) ): return value return json.loads(value) @abstractmethod def __call__(self) -> dict[str, Any]: pass class ConfigFileSourceMixin(ABC): @staticmethod def _existing_files(files: ConfigFileSourceType | None) -> list[Path | Traversable]: """Return the given files that exist, with `~` expanded in native paths.""" if files is None: return [] if not isinstance(files, Sequence) or isinstance(files, str): files = [files] existing: list[Path | Traversable] = [] for file in files: file_path: Path | Traversable = Path(file) if isinstance(file, str) else file if isinstance(file_path, Path): file_path = file_path.expanduser() if file_path.is_file(): existing.append(file_path) return existing def _read_files(self, files: ConfigFileSourceType | None, deep_merge: bool = False) -> dict[str, Any]: vars: dict[str, Any] = {} for file_path in self._existing_files(files): updating_vars = self._read_file(file_path) if deep_merge: vars = deep_update(vars, updating_vars) else: vars.update(updating_vars) return vars @abstractmethod def _read_file(self, path: Path | Traversable) -> dict[str, Any]: pass def _unwrap_optional_annotation(annotation: Any) -> Any: """Return the inner type of an ``Optional[T]`` annotation, otherwise the annotation unchanged.""" if is_union_origin(get_origin(annotation)): args = get_args(annotation) if len(args) == 2 and type(None) in args: for arg in args: # pragma: no branch if arg is not type(None): return arg return annotation def _is_model_or_dataclass(annotation: Any) -> bool: """Whether the annotation is a pydantic model or a pydantic dataclass, i.e. a type with nested fields.""" return is_model_class(annotation) or is_pydantic_dataclass(annotation) def _get_discriminator(field_info: FieldInfo) -> str | Callable[[Any], Hashable] | None: """Get the discriminator of a field using a discriminated union (via Annotated or Field(discriminator=...)).""" from pydantic import Discriminator discriminator = field_info.discriminator if discriminator is None: discriminator = next((m for m in field_info.metadata if isinstance(m, Discriminator)), None) return discriminator.discriminator if isinstance(discriminator, Discriminator) else discriminator def _get_member_fields(model_cls: type[Any]) -> dict[str, FieldInfo]: """Get the fields of a discriminated union member: a model, a pydantic dataclass or a stdlib dataclass.""" if _is_model_or_dataclass(model_cls): return _get_model_fields(model_cls) # Pydantic doesn't attach `__pydantic_fields__` to a stdlib dataclass, so collect the `FieldInfo`s itself: # `Field()` is kept as the dataclass field's default, or in its `Annotated` metadata. fields: dict[str, FieldInfo] = {} hints = get_type_hints(model_cls, include_extras=True) for dc_field in dataclass_fields(model_cls): if isinstance(dc_field.default, FieldInfo): fields[dc_field.name] = dc_field.default else: metadata = get_args(hints.get(dc_field.name))[1:] field_info = next((m for m in metadata if isinstance(m, FieldInfo)), None) if field_info is not None: fields[dc_field.name] = field_info return fields def _get_discriminator_keys(model_cls: type[Any], discriminator: str) -> tuple[str, ...]: """Get the keys that can hold the discriminator value of a discriminated union member: its name and alias.""" alias = getattr(_get_member_fields(model_cls).get(discriminator), 'validation_alias', None) return (discriminator, alias) if isinstance(alias, str) else (discriminator,) def _get_nested_discriminators(annotation: Any) -> list[str] | None: """Get the discriminators nested inside a discriminated union's annotation. A member can be guarded by more than one discriminator, e.g. in ``Annotated[Annotated[BlackCat | WhiteCat, Discriminator('color')] | Dog, Discriminator('pet_type')]`` reaching ``BlackCat`` requires both ``pet_type='cat'`` and ``color='black'``. Each one has to keep the default's value for its default to still apply. Returns ``None`` if a nested discriminator can't be resolved to a field name (i.e. it is callable), as the member it selects can't be looked up in the incoming value. """ from pydantic import Discriminator discriminators: list[str] = [] for arg in get_args(annotation): if arg is type(None): continue for metadata in get_args(arg)[1:]: if isinstance(metadata, Discriminator): if not isinstance(metadata.discriminator, str): return None discriminators.append(metadata.discriminator) nested = _get_nested_discriminators(arg) if nested is None: return None discriminators.extend(nested) return discriminators class DefaultSettingsSource(PydanticBaseSettingsSource): """ Source class for loading default object values. Args: settings_cls: The Settings class. nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields. Defaults to `False`. """ def __init__( self, settings_cls: type[BaseSettings], nested_model_default_partial_update: bool | None = None, _init_state: InitState | None = None, ): super().__init__(settings_cls, _init_state) self.defaults: dict[str, Any] = {} # Discriminated union fields: per discriminator guarding the default's member, the keys that can hold # its value, and the default's value self._discriminator_tags: dict[str, list[tuple[tuple[str, ...], Any]]] = {} self.nested_model_default_partial_update = ( nested_model_default_partial_update if nested_model_default_partial_update is not None else self.config.get('nested_model_default_partial_update', False) ) if self.nested_model_default_partial_update: validate_by_alias = _validate_by_alias_enabled(self.config) for field_name, field_info in settings_cls.model_fields.items(): _warn_if_field_info_incomplete(field_info, field_name, self._init_state) discriminator = _get_discriminator(field_info) if callable(discriminator): # The member selected by a callable discriminator can't be looked up in the incoming value continue nested = _get_nested_discriminators(field_info.annotation) if discriminator is not None else [] if nested is None: # Same for a nested callable discriminator continue alias_names, *_ = _get_alias_names(field_name, field_info, validate_by_alias=validate_by_alias) preferred_alias = alias_names[0] if is_dataclass(type(field_info.default)): self.defaults[preferred_alias] = asdict(field_info.default) elif is_model_class(type(field_info.default)): self.defaults[preferred_alias] = field_info.default.model_dump() if discriminator is not None and preferred_alias in self.defaults: default_cls = type(field_info.default) self._discriminator_tags[preferred_alias] = [ (_get_discriminator_keys(default_cls, name), getattr(field_info.default, name, None)) for name in (discriminator, *nested) ] def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: # Nothing to do here. Only implement the return statement to make mypy happy return None, '', False def __call__(self) -> dict[str, Any]: defaults = self.defaults.copy() for alias, discriminators in self._discriminator_tags.items(): value = self.current_state.get(alias) if not isinstance(value, dict): continue # Pydantic reads each tag from the discriminator field's name or its alias, so a value under any of # those keys selects a member. A key that is absent leaves the default's member in place. # The default's fields don't apply when any discriminator selects another member. if any(value[key] != default_tag for keys, default_tag in discriminators for key in keys if key in value): del defaults[alias] return defaults def __repr__(self) -> str: return ( f'{self.__class__.__name__}(nested_model_default_partial_update={self.nested_model_default_partial_update})' ) class InitSettingsSource(PydanticBaseSettingsSource): """ Source class for loading values provided during settings class initialization. """ def __init__( self, settings_cls: type[BaseSettings], init_kwargs: dict[str, Any], nested_model_default_partial_update: bool | None = None, _init_state: InitState | None = None, ): super().__init__(settings_cls, _init_state) case_sensitive = self.config.get('case_sensitive', False) include_name = _validate_by_name_enabled(self.config) validate_by_alias = _validate_by_alias_enabled(self.config) def normalize(name: str) -> str: # When case_sensitive is False, matching is done on lower-cased names. return name if case_sensitive else name.lower() self.init_kwargs = {} init_kwarg_names = set(init_kwargs.keys()) # Map each normalized name to all provided keys sharing it, so field aliases can # be matched without losing the original key used to look up the value. A 1-to-many # mapping is required because case_sensitive=False can collapse distinct keys # (e.g. ``TeSt`` and ``TEST``) onto the same normalized name. normalized_provided: dict[str, list[str]] = {} for key in init_kwargs: normalized_provided.setdefault(normalize(key), []).append(key) for field_name, field_info in settings_cls.model_fields.items(): _warn_if_field_info_incomplete(field_info, field_name, self._init_state) # Canonical (case-preserving) alias names determine the output key, so the value # still matches pydantic's validation aliases, which are always case-sensitive. canonical_alias_names, *_ = _get_alias_names(field_name, field_info, validate_by_alias=validate_by_alias) # Match alias names respect case_sensitive when matching against provided keys. match_alias_names, *_ = _get_alias_names( field_name, field_info, case_sensitive=case_sensitive, validate_by_alias=validate_by_alias ) # When populate_by_name is True, allow using the field name as an input key, # but normalize to the preferred alias to keep keys consistent across sources. matchable_names = list(match_alias_names) if include_name: name = normalize(field_name) if name not in matchable_names: matchable_names.append(name) # All provided keys matching this field, in matchable_names preference order # (aliases first, then the field name). Only keys not yet consumed by an # earlier field are considered. matched_keys = [ original for name in matchable_names if name in normalized_provided for original in normalized_provided[name] if original in init_kwarg_names ] if matched_keys: preferred_alias = canonical_alias_names[0] if canonical_alias_names else field_name # Prefer the first match in preference order; drop the remaining matches # so they are not re-added as extras. init_kwarg_names.difference_update(matched_keys) self.init_kwargs[preferred_alias] = init_kwargs[matched_keys[0]] # Include any remaining init kwargs (e.g., extras) unchanged # Note: If populate_by_name is True and the provided key is the field name, but # no alias exists, we keep it as-is so it can be processed as extra if allowed. self.init_kwargs.update({key: val for key, val in init_kwargs.items() if key in init_kwarg_names}) self.nested_model_default_partial_update = ( nested_model_default_partial_update if nested_model_default_partial_update is not None else self.config.get('nested_model_default_partial_update', False) ) def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: # Nothing to do here. Only implement the return statement to make mypy happy return None, '', False def __call__(self) -> dict[str, Any]: return ( TypeAdapter(dict[str, Any]).dump_python(self.init_kwargs) if self.nested_model_default_partial_update else self.init_kwargs ) def __repr__(self) -> str: return f'{self.__class__.__name__}(init_kwargs={self.init_kwargs!r})' class PydanticBaseEnvSettingsSource(PydanticBaseSettingsSource): def __init__( self, settings_cls: type[BaseSettings], case_sensitive: bool | None = None, env_prefix: str | None = None, env_prefix_target: EnvPrefixTarget | None = None, env_ignore_empty: bool | None = None, env_parse_none_str: str | None = None, env_parse_enums: bool | None = None, _init_state: InitState | None = None, ) -> None: super().__init__(settings_cls, _init_state) self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False) self.env_prefix = env_prefix if env_prefix is not None else self.config.get('env_prefix', '') self.env_prefix_target = ( env_prefix_target if env_prefix_target is not None else self.config.get('env_prefix_target', 'variable') ) self.env_ignore_empty = ( env_ignore_empty if env_ignore_empty is not None else self.config.get('env_ignore_empty', False) ) self.env_parse_none_str = ( 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') def _apply_case_sensitive(self, value: str) -> str: return value.lower() if not self.case_sensitive else value def _extract_field_info( self, field: FieldInfo, field_name: str, model_config: Mapping[str, Any] | None = None ) -> list[tuple[str, str, bool]]: """ Extracts field info. This info is used to get the value of field from environment variables. It returns a list of tuples, each tuple contains: * field_key: The key of field that has to be used in model creation. * env_name: The environment variable name of the field. * value_is_complex: A flag to determine whether the value from environment variable is complex and has to be parsed. Args: field (FieldInfo): The field. field_name (str): The field name. model_config (Mapping[str, Any] | None): The config of the model owning the field, which decides whether its alias and name are accepted. Defaults to the settings config, pass it for a field of a nested model. Returns: list[tuple[str, str, bool]]: List of tuples, each tuple contains field_key, env_name, and value_is_complex. """ _warn_if_field_info_incomplete(field, field_name, self._init_state) field_info: list[tuple[str, str, bool]] = [] v_alias: str | list[str | int] | list[list[str | int]] | None if model_config is None: model_config = self.config if not _validate_by_alias_enabled(model_config): # pydantic ignores the aliases, so only the field name can populate the field. v_alias = None elif isinstance(field.validation_alias, (AliasChoices, AliasPath)): v_alias = field.validation_alias.convert_to_aliases() else: v_alias = field.validation_alias if v_alias: env_prefix = self.env_prefix if self.env_prefix_target in ('alias', 'all') else '' if isinstance(v_alias, list): # AliasChoices, AliasPath if isinstance(v_alias[0], list): # AliasChoices: one entry per choice for alias in v_alias: first_arg = cast(str, alias[0]) # first item of an AliasChoices must be a str field_info.append( ( first_arg, self._apply_case_sensitive(env_prefix + first_arg), len(alias) > 1, ) ) else: # bare AliasPath: a single flat path, e.g. ['a', 0, 'b'] # Only the head names the env var to read; the remaining segments are # navigated in-memory once the value has been JSON-decoded, so they must # never be looked up as env vars of their own (previously each string # segment produced its own candidate). Complexity must reflect the number # of segments in the whole path, not the length of the head's name. head = cast(str, v_alias[0]) # first item of an AliasPath must be a str field_info.append((head, self._apply_case_sensitive(env_prefix + head), len(v_alias) > 1)) else: # string validation alias field_info.append((v_alias, self._apply_case_sensitive(env_prefix + v_alias), False)) if not v_alias or _validate_by_name_enabled(model_config): annotation, metadata = _get_field_annotation_and_metadata(field) env_prefix = self.env_prefix if self.env_prefix_target in ('variable', 'all') else '' if is_union_origin(get_origin(annotation)) and _union_is_complex(annotation, metadata, self._init_state): field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), True)) else: field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), False)) return field_info def _replace_field_names_case_insensitively(self, field: FieldInfo, field_values: dict[str, Any]) -> dict[str, Any]: """ Replace field names in values dict by looking in models fields insensitively. By having the following models: ```py class SubSubSub(BaseModel): VaL3: str class SubSub(BaseModel): Val2: str SUB_sub_SuB: SubSubSub class Sub(BaseModel): VAL1: str SUB_sub: SubSub class Settings(BaseSettings): nested: Sub model_config = SettingsConfigDict(env_nested_delimiter='__') ``` Then: _replace_field_names_case_insensitively( field, {"val1": "v1", "sub_SUB": {"VAL2": "v2", "sub_SUB_sUb": {"vAl3": "v3"}}} ) Returns {'VAL1': 'v1', 'SUB_sub': {'Val2': 'v2', 'SUB_sub_SuB': {'VaL3': 'v3'}}} """ values: dict[str, Any] = {} for name, value in field_values.items(): sub_model_field: FieldInfo | None = None annotation = _strip_annotated(_unwrap_optional_annotation(field.annotation)) if not _is_model_or_dataclass(annotation): values[name] = value continue model_fields = _get_model_fields(annotation) validate_by_alias = _validate_by_alias_enabled(_get_model_config(annotation)) # Find field in sub model by looking in fields case insensitively field_key: str | None = None for sub_model_field_name, sub_model_field in model_fields.items(): _warn_if_field_info_incomplete(sub_model_field, sub_model_field_name, self._init_state) aliases, _ = _get_alias_names( sub_model_field_name, sub_model_field, validate_by_alias=validate_by_alias ) _search = (alias for alias in aliases if alias.lower() == name.lower()) if field_key := next(_search, None): break if not field_key: values[name] = value continue if ( sub_model_field is not None and _is_model_or_dataclass(_strip_annotated(_unwrap_optional_annotation(sub_model_field.annotation))) and isinstance(value, dict) ): values[field_key] = self._replace_field_names_case_insensitively(sub_model_field, value) else: values[field_key] = value return values def _replace_env_none_type_values(self, field_value: dict[str, Any]) -> dict[str, Any]: """ Recursively parse values that are of "None" type(EnvNoneType) to `None` type(None). """ values: dict[str, Any] = {} for key, value in field_value.items(): if not isinstance(value, EnvNoneType): values[key] = value if not isinstance(value, dict) else self._replace_env_none_type_values(value) else: values[key] = None return values def _get_resolved_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: """ Gets the value, the preferred alias key for model creation, and a flag to determine whether value is complex. Note: In V3, this method should either be made public, or, this method should be removed and the abstract method get_field_value should be updated to include a "use_preferred_alias" flag. Args: field: The field. field_name: The field name. Returns: A tuple that contains the value, preferred key and a flag to determine whether value is complex. """ field_value, field_key, value_is_complex = self.get_field_value(field, field_name) if not (value_is_complex or (_validate_by_name_enabled(self.config) and (field_key == field_name))): field_infos = self._extract_field_info(field, field_name) preferred_key, _, preferred_is_complex = field_infos[0] # Only normalize to preferred_key when it's a simple string alias. # When the preferred key comes from an AliasPath (complex entry), skip normalization # to avoid using the AliasPath's first element as the key (see #766). if not preferred_is_complex: return field_value, preferred_key, value_is_complex return field_value, field_key, value_is_complex def __call__(self) -> dict[str, Any]: data: dict[str, Any] = {} for field_name, field in self.settings_cls.model_fields.items(): _warn_if_field_info_incomplete(field, field_name, self._init_state) try: field_value, field_key, value_is_complex = self._get_resolved_field_value(field, field_name) except Exception as e: raise SettingsError( f'error getting value for field "{field_name}" from source "{self.__class__.__name__}"' ) from e try: field_value = self.prepare_field_value(field_name, field, field_value, value_is_complex) except ValueError as e: raise SettingsError( f'error parsing value for field "{field_name}" from source "{self.__class__.__name__}"' ) from e if field_value is not None: if self.env_parse_none_str is not None: if isinstance(field_value, dict): field_value = self._replace_env_none_type_values(field_value) elif isinstance(field_value, EnvNoneType): field_value = None if ( not self.case_sensitive # and _lenient_issubclass(field.annotation, BaseModel) and isinstance(field_value, dict) ): data[field_key] = self._replace_field_names_case_insensitively(field, field_value) else: data[field_key] = field_value return data __all__ = [ 'ConfigFileSourceMixin', 'DefaultSettingsSource', 'InitSettingsSource', 'PydanticBaseEnvSettingsSource', 'PydanticBaseSettingsSource', 'SettingsError', ]