Загрузить файлы в «venv/Lib/site-packages/pydantic_settings/sources»
This commit is contained in:
86
venv/Lib/site-packages/pydantic_settings/sources/__init__.py
Normal file
86
venv/Lib/site-packages/pydantic_settings/sources/__init__.py
Normal file
@@ -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',
|
||||
]
|
||||
582
venv/Lib/site-packages/pydantic_settings/sources/base.py
Normal file
582
venv/Lib/site-packages/pydantic_settings/sources/base.py
Normal file
@@ -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',
|
||||
]
|
||||
100
venv/Lib/site-packages/pydantic_settings/sources/types.py
Normal file
100
venv/Lib/site-packages/pydantic_settings/sources/types.py
Normal file
@@ -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',
|
||||
]
|
||||
330
venv/Lib/site-packages/pydantic_settings/sources/utils.py
Normal file
330
venv/Lib/site-packages/pydantic_settings/sources/utils.py
Normal file
@@ -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',
|
||||
]
|
||||
Reference in New Issue
Block a user