From e18aa491d028abfb3b41a065ccf8343faf7804f6 Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:11:47 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/pydantic=5Fsettings/sources=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../pydantic_settings/sources/__init__.py | 86 +++ .../pydantic_settings/sources/base.py | 582 ++++++++++++++++++ .../pydantic_settings/sources/types.py | 100 +++ .../pydantic_settings/sources/utils.py | 330 ++++++++++ 4 files changed, 1098 insertions(+) create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/__init__.py create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/base.py create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/types.py create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/utils.py diff --git a/venv/Lib/site-packages/pydantic_settings/sources/__init__.py b/venv/Lib/site-packages/pydantic_settings/sources/__init__.py new file mode 100644 index 0000000..96dbf36 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/__init__.py @@ -0,0 +1,86 @@ +"""Package for handling configuration sources in pydantic-settings.""" + +from .base import ( + ConfigFileSourceMixin, + DefaultSettingsSource, + InitSettingsSource, + PydanticBaseEnvSettingsSource, + PydanticBaseSettingsSource, + get_subcommand, +) +from .providers.aws import AWSSecretsManagerSettingsSource +from .providers.azure import AzureKeyVaultSettingsSource +from .providers.cli import ( + CLI_SUPPRESS, + CliDualFlag, + CliExplicitFlag, + CliImplicitFlag, + CliMutuallyExclusiveGroup, + CliPositionalArg, + CliSettingsSource, + CliSubCommand, + CliSuppress, + CliToggleFlag, + CliUnknownArgs, +) +from .providers.dotenv import DotEnvSettingsSource, read_env_file +from .providers.env import EnvSettingsSource +from .providers.gcp import GoogleSecretManagerSettingsSource +from .providers.json import JsonConfigSettingsSource +from .providers.nested_secrets import NestedSecretsSettingsSource +from .providers.pyproject import PyprojectTomlConfigSettingsSource +from .providers.secrets import SecretsSettingsSource +from .providers.toml import TomlConfigSettingsSource +from .providers.yaml import YamlConfigSettingsSource +from .types import ( + DEFAULT_PATH, + ENV_FILE_SENTINEL, + DotenvFiltering, + DotenvType, + EnvPrefixTarget, + ForceDecode, + NoDecode, + PathType, + PydanticModel, +) + +__all__ = [ + 'CLI_SUPPRESS', + 'ENV_FILE_SENTINEL', + 'DEFAULT_PATH', + 'AWSSecretsManagerSettingsSource', + 'AzureKeyVaultSettingsSource', + 'CliExplicitFlag', + 'CliImplicitFlag', + 'CliToggleFlag', + 'CliDualFlag', + 'CliMutuallyExclusiveGroup', + 'CliPositionalArg', + 'CliSettingsSource', + 'CliSubCommand', + 'CliSuppress', + 'CliUnknownArgs', + 'DefaultSettingsSource', + 'DotEnvSettingsSource', + 'DotenvFiltering', + 'DotenvType', + 'EnvPrefixTarget', + 'EnvSettingsSource', + 'ForceDecode', + 'GoogleSecretManagerSettingsSource', + 'InitSettingsSource', + 'JsonConfigSettingsSource', + 'NestedSecretsSettingsSource', + 'NoDecode', + 'PathType', + 'PydanticBaseEnvSettingsSource', + 'PydanticBaseSettingsSource', + 'ConfigFileSourceMixin', + 'PydanticModel', + 'PyprojectTomlConfigSettingsSource', + 'SecretsSettingsSource', + 'TomlConfigSettingsSource', + 'YamlConfigSettingsSource', + 'get_subcommand', + 'read_env_file', +] diff --git a/venv/Lib/site-packages/pydantic_settings/sources/base.py b/venv/Lib/site-packages/pydantic_settings/sources/base.py new file mode 100644 index 0000000..eafe713 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/base.py @@ -0,0 +1,582 @@ +"""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 Sequence +from dataclasses import asdict, is_dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any, cast, get_args + +from pydantic import AliasChoices, AliasPath, BaseModel, 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.fields import FieldInfo +from typing_inspection.introspection import is_union_origin + +from ..exceptions import SettingsError +from ..utils import _lenient_issubclass +from .types import EnvNoneType, EnvPrefixTarget, ForceDecode, NoDecode, PathType, PydanticModel, _CliSubCommand +from .utils import ( + _annotation_is_complex, + _get_alias_names, + _get_field_metadata, + _get_model_fields, + _resolve_type_alias, + _strip_annotated, + _union_is_complex, +) + +if TYPE_CHECKING: + from pydantic_settings.main import BaseSettings + + +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) + + 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]): + 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. + """ + pass + + 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) + + 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 + """ + if field and ( + NoDecode in _get_field_metadata(field) + or (self.config.get('enable_decoding') is False and ForceDecode not in field.metadata) + ): + return value + + return json.loads(value) + + @abstractmethod + def __call__(self) -> dict[str, Any]: + pass + + +class ConfigFileSourceMixin(ABC): + def _read_files(self, files: PathType | None, deep_merge: bool = False) -> dict[str, Any]: + if files is None: + return {} + if not isinstance(files, Sequence) or isinstance(files, str): + files = [files] + vars: dict[str, Any] = {} + for file in files: + if isinstance(file, str): + file_path = Path(file) + else: + file_path = file + if isinstance(file_path, Path): + file_path = file_path.expanduser() + + if not file_path.is_file(): + continue + + 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) -> dict[str, Any]: + pass + + +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): + super().__init__(settings_cls) + self.defaults: dict[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: + for field_name, field_info in settings_cls.model_fields.items(): + alias_names, *_ = _get_alias_names(field_name, field_info) + 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() + + 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 self.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, + ): + self.init_kwargs = {} + init_kwarg_names = set(init_kwargs.keys()) + for field_name, field_info in settings_cls.model_fields.items(): + alias_names, *_ = _get_alias_names(field_name, field_info) + # 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 = set(alias_names) + include_name = settings_cls.model_config.get('populate_by_name', False) or settings_cls.model_config.get( + 'validate_by_name', False + ) + if include_name: + matchable_names.add(field_name) + init_kwarg_name = init_kwarg_names & matchable_names + if init_kwarg_name: + preferred_alias = alias_names[0] if alias_names else field_name + # Choose provided key deterministically: prefer the first alias in alias_names order; + # fall back to field_name if allowed and provided. + provided_key = next((alias for alias in alias_names if alias in init_kwarg_names), None) + if provided_key is None and include_name and field_name in init_kwarg_names: + provided_key = field_name + # provided_key should not be None here because init_kwarg_name is non-empty + assert provided_key is not None + init_kwarg_names -= init_kwarg_name + self.init_kwargs[preferred_alias] = init_kwargs[provided_key] + # 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}) + + super().__init__(settings_cls) + 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, + ) -> None: + super().__init__(settings_cls) + 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) -> 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. + + Returns: + list[tuple[str, str, bool]]: List of tuples, each tuple contains field_key, env_name, and value_is_complex. + """ + field_info: list[tuple[str, str, bool]] = [] + if isinstance(field.validation_alias, (AliasChoices, AliasPath)): + v_alias: str | list[str | int] | list[list[str | int]] | None = 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 + for alias in v_alias: + if isinstance(alias, str): # AliasPath + field_info.append( + (alias, self._apply_case_sensitive(env_prefix + alias), True if len(alias) > 1 else False) + ) + elif isinstance(alias, list): # AliasChoices + 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), + True if len(alias) > 1 else False, + ) + ) + else: # string validation alias + field_info.append((v_alias, self._apply_case_sensitive(env_prefix + v_alias), False)) + + if not v_alias or self.config.get('populate_by_name', False) or self.config.get('validate_by_name', False): + annotation = _strip_annotated(_resolve_type_alias(field.annotation)) + 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, field.metadata): + 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 = field.annotation + + # If field is Optional, we need to find the actual type + if is_union_origin(get_origin(field.annotation)): + args = get_args(annotation) + if len(args) == 2 and type(None) in args: + for arg in args: + if arg is not None: + annotation = arg + break + + # This is here to make mypy happy + # Item "None" of "Optional[Type[Any]]" has no attribute "model_fields" + if not annotation or not hasattr(annotation, 'model_fields'): + values[name] = value + continue + else: + model_fields: dict[str, FieldInfo] = annotation.model_fields + + # 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(): + aliases, _ = _get_alias_names(sub_model_field_name, sub_model_field) + _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 _lenient_issubclass(sub_model_field.annotation, BaseModel) + 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 ( + (self.config.get('populate_by_name', False) or self.config.get('validate_by_name', False)) + 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(): + 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', +] diff --git a/venv/Lib/site-packages/pydantic_settings/sources/types.py b/venv/Lib/site-packages/pydantic_settings/sources/types.py new file mode 100644 index 0000000..61d5165 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/types.py @@ -0,0 +1,100 @@ +"""Type definitions for pydantic-settings sources.""" + +from __future__ import annotations as _annotations + +from collections.abc import Sequence +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal + +if TYPE_CHECKING: + from pydantic._internal._dataclasses import PydanticDataclass + from pydantic.main import BaseModel + + PydanticModel = PydanticDataclass | BaseModel +else: + PydanticModel = Any + + +class EnvNoneType(str): + pass + + +class NoDecode: + """Annotation to prevent decoding of a field value.""" + + pass + + +class ForceDecode: + """Annotation to force decoding of a field value.""" + + pass + + +EnvPrefixTarget = Literal['variable', 'alias', 'all'] +DotenvType = Path | str | Sequence[Path | str] +PathType = Path | str | Sequence[Path | str] +DotenvFiltering = Literal['match_prefix', 'only_existing'] +DEFAULT_PATH: PathType = Path('') + +# This is used as default value for `_env_file` in the `BaseSettings` class and +# `env_file` in `DotEnvSettingsSource` so the default can be distinguished from `None`. +# See the docstring of `BaseSettings` for more details. +ENV_FILE_SENTINEL: DotenvType = Path('') + + +class _CliSubCommand: + pass + + +class _CliPositionalArg: + pass + + +class _CliImplicitFlag: + pass + + +class _CliToggleFlag(_CliImplicitFlag): + pass + + +class _CliDualFlag(_CliImplicitFlag): + pass + + +class _CliExplicitFlag: + pass + + +class _CliUnknownArgs: + pass + + +class SecretVersion: + def __init__(self, version: str) -> None: + self.version = version + + def __repr__(self) -> str: + return f'{self.__class__.__name__}({self.version!r})' + + +__all__ = [ + 'DEFAULT_PATH', + 'ENV_FILE_SENTINEL', + 'EnvPrefixTarget', + 'DotenvType', + 'EnvNoneType', + 'ForceDecode', + 'NoDecode', + 'PathType', + 'PydanticModel', + 'SecretVersion', + '_CliExplicitFlag', + '_CliImplicitFlag', + '_CliToggleFlag', + '_CliDualFlag', + '_CliPositionalArg', + '_CliSubCommand', + '_CliUnknownArgs', +] diff --git a/venv/Lib/site-packages/pydantic_settings/sources/utils.py b/venv/Lib/site-packages/pydantic_settings/sources/utils.py new file mode 100644 index 0000000..10c79ad --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/utils.py @@ -0,0 +1,330 @@ +"""Utility functions for pydantic-settings sources.""" + +from __future__ import annotations as _annotations + +from collections import deque +from collections.abc import Mapping, Sequence +from dataclasses import is_dataclass +from enum import Enum +from typing import Any, TypeVar, cast, get_args, get_origin + +from pydantic import BaseModel, Json, RootModel, Secret +from pydantic._internal._utils import is_model_class +from pydantic.dataclasses import is_pydantic_dataclass +from pydantic.fields import FieldInfo +from pydantic.types import Strict +from typing_inspection import typing_objects +from typing_inspection.introspection import is_union_origin + +from ..exceptions import SettingsError +from ..utils import _lenient_issubclass +from .types import EnvNoneType + + +def _get_env_var_key(key: str, case_sensitive: bool = False) -> str: + return key if case_sensitive else key.lower() + + +def _parse_env_none_str(value: str | None, parse_none_str: str | None = None) -> str | None | EnvNoneType: + return value if not (value == parse_none_str and parse_none_str is not None) else EnvNoneType(value) + + +def parse_env_vars( + env_vars: Mapping[str, str | None], + case_sensitive: bool = False, + ignore_empty: bool = False, + parse_none_str: str | None = None, +) -> Mapping[str, str | None]: + return { + _get_env_var_key(k, case_sensitive): _parse_env_none_str(v, parse_none_str) + for k, v in env_vars.items() + if not (ignore_empty and v == '') + } + + +def _substitute_typevars(tp: Any, param_map: dict[Any, Any]) -> Any: + """Substitute TypeVars in a type annotation with concrete types from param_map.""" + if isinstance(tp, TypeVar) and tp in param_map: + return param_map[tp] + args = get_args(tp) + if not args: + return tp + new_args = tuple(_substitute_typevars(arg, param_map) for arg in args) + if new_args == args: + return tp + origin = get_origin(tp) + if origin is not None: + try: + return origin[new_args] + except TypeError: + # types.UnionType and similar are not directly subscriptable, + # reconstruct using | operator + import functools + import operator + + return functools.reduce(operator.or_, new_args) + return tp + + +def _resolve_type_alias(annotation: Any) -> Any: + """Resolve a TypeAliasType to its underlying value, substituting type params if parameterized.""" + if typing_objects.is_typealiastype(annotation): + return annotation.__value__ + origin = get_origin(annotation) + if typing_objects.is_typealiastype(origin): + type_params = getattr(origin, '__type_params__', ()) + type_args = get_args(annotation) + value = origin.__value__ + if type_params and type_args: + return _substitute_typevars(value, dict(zip(type_params, type_args))) + return value + return annotation + + +def _annotation_is_complex(annotation: Any, metadata: list[Any]) -> bool: + # If the model is a root model, the root annotation should be used to + # evaluate the complexity. + annotation = _resolve_type_alias(annotation) + if annotation is not None and _lenient_issubclass(annotation, RootModel) and annotation is not RootModel: + annotation = cast('type[RootModel[Any]]', annotation) + root_annotation = annotation.model_fields['root'].annotation + if root_annotation is not None: # pragma: no branch + annotation = root_annotation + + if any(isinstance(md, Json) for md in metadata): # type: ignore[misc] + return False + + origin = get_origin(annotation) + + # Check if annotation is of the form Annotated[type, metadata]. + if typing_objects.is_annotated(origin): + # Return result of recursive call on inner type. + inner, *meta = get_args(annotation) + return _annotation_is_complex(inner, meta) + + if origin is Secret: + return False + + return ( + _annotation_is_complex_inner(annotation) + or _annotation_is_complex_inner(origin) + or hasattr(origin, '__pydantic_core_schema__') + or hasattr(origin, '__get_pydantic_core_schema__') + ) + + +def _get_field_metadata(field: FieldInfo) -> list[Any]: + annotation = _resolve_type_alias(field.annotation) + metadata = field.metadata + origin = get_origin(annotation) + if typing_objects.is_annotated(origin): + _, *meta = get_args(annotation) + metadata += meta + return metadata + + +def _annotation_is_complex_inner(annotation: type[Any] | None) -> bool: + if _lenient_issubclass(annotation, (str, bytes)): + return False + + return _lenient_issubclass( + annotation, (BaseModel, Mapping, Sequence, tuple, set, frozenset, deque) + ) or is_dataclass(annotation) + + +def _union_is_complex(annotation: type[Any] | None, metadata: list[Any]) -> bool: + """Check if a union type contains any complex types.""" + for arg in get_args(annotation): + if _annotation_is_complex(arg, metadata): + return True + # _annotation_is_complex doesn't handle bare Union types, so when an arg + # is Annotated[Union[X, Y], ...], stripping Annotated yields a bare Union + # that _annotation_is_complex can't evaluate. Recurse into it, but only + # if the Annotated metadata doesn't suppress complexity (e.g. Json). + inner = _strip_annotated(arg) + if inner is not arg: + _, *inner_meta = get_args(arg) + if any(isinstance(md, Json) for md in inner_meta): # type: ignore[misc] + continue + if is_union_origin(get_origin(inner)): + if _union_is_complex(inner, metadata): + return True + return False + + +def _union_has_strict_types(annotation: type[Any] | None) -> bool: + """Check if a union type contains any strict-annotated types.""" + for arg in get_args(annotation): + if typing_objects.is_annotated(get_origin(arg)): + _, *meta = get_args(arg) + if any(isinstance(m, Strict) for m in meta): + return True + return False + + +def _annotation_contains_types( + annotation: type[Any] | None, + types: tuple[Any, ...], + is_include_origin: bool = True, + is_strip_annotated: bool = False, + is_instance: bool = False, + collect: set[Any] | None = None, +) -> bool: + """Check if a type annotation contains any of the specified types.""" + if is_strip_annotated: + annotation = _strip_annotated(annotation) + if is_include_origin is True: + origin = get_origin(annotation) + if origin in types: + if collect is None: + return True + collect.add(annotation) + if is_instance and any(isinstance(origin, type_) for type_ in types): + if collect is None: + return True + collect.add(annotation) + for type_ in get_args(annotation): + if ( + _annotation_contains_types( + type_, + types, + is_include_origin=True, + is_strip_annotated=is_strip_annotated, + is_instance=is_instance, + collect=collect, + ) + and collect is None + ): + return True + if is_instance and any(isinstance(annotation, type_) for type_ in types): + if collect is None: + return True + collect.add(annotation) + if annotation in types: + if collect is not None: + collect.add(annotation) + return True + return False + + +def _strip_annotated(annotation: Any) -> Any: + if typing_objects.is_annotated(get_origin(annotation)): + return annotation.__origin__ + else: + return annotation + + +def _annotation_enum_val_to_name(annotation: type[Any] | None, value: Any) -> str | None: + for type_ in (annotation, get_origin(annotation), *get_args(annotation)): + if _lenient_issubclass(type_, Enum): + if value in type_.__members__.values(): + return type_(value).name + return None + + +def _annotation_enum_name_to_val(annotation: type[Any] | None, name: Any) -> Any: + for type_ in (annotation, get_origin(annotation), *get_args(annotation)): + if _lenient_issubclass(type_, Enum): + if name in type_.__members__.keys(): + return type_[name] + return None + + +def _literal_has_numeric_enum(annotation: type[Any] | None) -> bool: + """Check if annotation is a Literal type containing numeric Enum members (IntEnum, (int, Enum), (float, Enum)).""" + if typing_objects.is_literal(get_origin(annotation)): + return any(isinstance(arg, (int, float)) and isinstance(arg, Enum) for arg in get_args(annotation)) + # Handle Annotated wrapping, e.g. Annotated[Literal[IntEnum.member], Field(...)] + if typing_objects.is_annotated(get_origin(annotation)): + inner = get_args(annotation)[0] + return _literal_has_numeric_enum(inner) + # Handle Union/Optional wrapping, e.g. Optional[Literal[IntEnum.member]] + if is_union_origin(get_origin(annotation)): + return any(_literal_has_numeric_enum(arg) for arg in get_args(annotation)) + return False + + +def _get_model_fields(model_cls: type[Any]) -> dict[str, Any]: + """Get fields from a pydantic model or dataclass.""" + + if is_pydantic_dataclass(model_cls) and hasattr(model_cls, '__pydantic_fields__'): + return model_cls.__pydantic_fields__ + if is_model_class(model_cls): + return model_cls.model_fields + raise SettingsError(f'Error: {model_cls.__name__} is not subclass of BaseModel or pydantic.dataclasses.dataclass') + + +def _get_alias_names( + field_name: str, + field_info: Any, + alias_path_args: dict[str, int | None] | None = None, + case_sensitive: bool = True, + populate_by_name: bool = False, +) -> tuple[tuple[str, ...], bool]: + """Get alias names for a field, handling alias paths and case sensitivity.""" + from pydantic import AliasChoices, AliasPath + + alias_names: list[str] = [] + is_alias_path_only: bool = True + if not any((field_info.alias, field_info.validation_alias)): + alias_names += [field_name] + is_alias_path_only = False + else: + new_alias_paths: list[AliasPath] = [] + for alias in (field_info.alias, field_info.validation_alias): + if alias is None: + continue + elif isinstance(alias, str): + alias_names.append(alias) + is_alias_path_only = False + elif isinstance(alias, AliasChoices): + for name in alias.choices: + if isinstance(name, str): + alias_names.append(name) + is_alias_path_only = False + else: + new_alias_paths.append(name) + else: + new_alias_paths.append(alias) + for alias_path in new_alias_paths: + name = cast(str, alias_path.path[0]) + name = name.lower() if not case_sensitive else name + if alias_path_args is not None: + alias_path_args[name] = ( + alias_path.path[1] if len(alias_path.path) > 1 and isinstance(alias_path.path[1], int) else None + ) + if not alias_names and is_alias_path_only: + alias_names.append(name) + if populate_by_name and field_name not in alias_names: + alias_names.append(field_name) + is_alias_path_only = False + if not case_sensitive: + alias_names = [alias_name.lower() for alias_name in alias_names] + return tuple(dict.fromkeys(alias_names)), is_alias_path_only + + +def _is_function(obj: Any) -> bool: + """Check if an object is a function.""" + from types import BuiltinFunctionType, FunctionType + + return isinstance(obj, (FunctionType, BuiltinFunctionType)) + + +__all__ = [ + '_annotation_contains_types', + '_annotation_enum_name_to_val', + '_annotation_enum_val_to_name', + '_annotation_is_complex', + '_annotation_is_complex_inner', + '_get_alias_names', + '_get_env_var_key', + '_get_model_fields', + '_is_function', + '_literal_has_numeric_enum', + '_parse_env_none_str', + '_resolve_type_alias', + '_strip_annotated', + '_union_has_strict_types', + '_union_is_complex', + 'parse_env_vars', +]