Merge branch 'main' into deep-merge

Author: pmeier

.github/workflows/ci.yml

@@ -36,7 +36,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        python: ['3.9', '3.10', '3.11', '3.12', '3.13']
+        python: ['3.10', '3.11', '3.12', '3.13', '3.14']
 
     env:
       PYTHON: ${{ matrix.python }}

pydantic_settings/main.py

@@ -35,6 +35,7 @@
     YamlConfigSettingsSource,
     get_subcommand,
 )
+from .sources.utils import _get_alias_names
 
 T = TypeVar('T')
 
@@ -444,13 +445,37 @@ def _settings_build_values(
 
             # Strip any default values not explicity set before returning final state
             state = {key: val for key, val in state.items() if key not in defaults or defaults[key] != val}
+            self._settings_restore_init_kwarg_names(self.__class__, init_kwargs, state)
 
             return state
         else:
             # no one should mean to do this, but I think returning an empty dict is marginally preferable
             # to an informative error and much better than a confusing error
             return {}
 
+    @staticmethod
+    def _settings_restore_init_kwarg_names(
+        settings_cls: type[BaseSettings], init_kwargs: dict[str, Any], state: dict[str, Any]
+    ) -> None:
+        """
+        Restore the init_kwarg key names to the final merged state dictionary.
+        """
+        if init_kwargs and state:
+            state_kwarg_names = set(state.keys())
+            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)
+                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
+                state_kwarg_name = state_kwarg_names & matchable_names
+                if init_kwarg_name and state_kwarg_name:
+                    state[init_kwarg_name.pop()] = state.pop(state_kwarg_name.pop())
+
     @staticmethod
     def _settings_warn_unused_config_keys(sources: tuple[object, ...], model_config: SettingsConfigDict) -> None:
         """

pydantic_settings/sources/base.py

@@ -7,15 +7,14 @@
 from abc import ABC, abstractmethod
 from dataclasses import asdict, is_dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Optional, cast
+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_extensions import get_args
 from typing_inspection import typing_objects
 from typing_inspection.introspection import is_union_origin
 
@@ -36,7 +35,7 @@
 
 def get_subcommand(
     model: PydanticModel, is_required: bool = True, cli_exit_on_error: bool | None = None
-) -> Optional[PydanticModel]:
+) -> PydanticModel | None:
     """
     Get the subcommand from a model.
 
@@ -269,7 +268,9 @@ def __init__(
             # 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)
+            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
@@ -371,7 +372,7 @@ def _extract_field_info(self, field: FieldInfo, field_name: str) -> list[tuple[s
             else:  # string validation alias
                 field_info.append((v_alias, self._apply_case_sensitive(v_alias), False))
 
-        if not v_alias or self.config.get('populate_by_name', False):
+        if not v_alias or self.config.get('populate_by_name', False) or self.config.get('validate_by_name', False):
             annotation = field.annotation
             if typing_objects.is_typealiastype(annotation) or typing_objects.is_typealiastype(get_origin(annotation)):
                 annotation = _strip_annotated(annotation.__value__)  # type: ignore[union-attr]
@@ -491,7 +492,13 @@ def _get_resolved_field_value(self, field: FieldInfo, field_name: str) -> tuple[
             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) and (field_key == 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, *_ = field_infos[0]
             return field_value, preferred_key, value_is_complex

pydantic_settings/sources/providers/aws.py

@@ -2,7 +2,7 @@
 
 import json
 from collections.abc import Mapping
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from ..utils import parse_env_vars
 from .env import EnvSettingsSource
@@ -57,7 +57,7 @@ def __init__(
             env_parse_enums=env_parse_enums,
         )
 
-    def _load_env_vars(self) -> Mapping[str, Optional[str]]:
+    def _load_env_vars(self) -> Mapping[str, str | None]:
         response = self._secretsmanager_client.get_secret_value(SecretId=self._secret_id)  # type: ignore
 
         return parse_env_vars(

pydantic_settings/sources/providers/azure.py

@@ -3,7 +3,7 @@
 from __future__ import annotations as _annotations
 
 from collections.abc import Iterator, Mapping
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from pydantic.alias_generators import to_snake
 from pydantic.fields import FieldInfo
@@ -37,7 +37,7 @@ def import_azure_key_vault() -> None:
         ) from e
 
 
-class AzureKeyVaultMapping(Mapping[str, Optional[str]]):
+class AzureKeyVaultMapping(Mapping[str, str | None]):
     _loaded_secrets: dict[str, str | None]
     _secret_client: SecretClient
     _secret_names: list[str]
@@ -121,7 +121,7 @@ def __init__(
             env_parse_enums=env_parse_enums,
         )
 
-    def _load_env_vars(self) -> Mapping[str, Optional[str]]:
+    def _load_env_vars(self) -> Mapping[str, str | None]:
         secret_client = SecretClient(vault_url=self._url, credential=self._credential)
         return AzureKeyVaultMapping(
             secret_client=secret_client,

pydantic_settings/sources/providers/cli.py

@@ -16,7 +16,7 @@
     _SubParsersAction,
 )
 from collections import defaultdict
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from enum import Enum
 from functools import cached_property
 from textwrap import dedent
@@ -25,14 +25,13 @@
     TYPE_CHECKING,
     Annotated,
     Any,
-    Callable,
     Generic,
     Literal,
     NoReturn,
-    Optional,
     TypeVar,
-    Union,
     cast,
+    get_args,
+    get_origin,
     overload,
 )
 
@@ -43,7 +42,6 @@
 from pydantic.dataclasses import is_pydantic_dataclass
 from pydantic.fields import FieldInfo
 from pydantic_core import PydanticUndefined
-from typing_extensions import get_args, get_origin
 from typing_inspection import typing_objects
 from typing_inspection.introspection import is_union_origin
 
@@ -95,21 +93,21 @@ class _CliArg(BaseModel):
     arg_prefix: str
     case_sensitive: bool
     hide_none_type: bool
-    kebab_case: Optional[Union[bool, Literal['all', 'no_enums']]]
-    enable_decoding: Optional[bool]
+    kebab_case: bool | Literal['all', 'no_enums'] | None
+    enable_decoding: bool | None
     env_prefix_len: int
     args: list[str] = []
     kwargs: dict[str, Any] = {}
 
     _alias_names: tuple[str, ...] = PrivateAttr(())
-    _alias_paths: dict[str, Optional[int]] = PrivateAttr({})
+    _alias_paths: dict[str, int | None] = PrivateAttr({})
     _is_alias_path_only: bool = PrivateAttr(False)
     _field_info: FieldInfo = PrivateAttr()
 
     def __init__(
         self,
         field_info: FieldInfo,
-        parser_map: defaultdict[str | FieldInfo, dict[Optional[int] | str, _CliArg]],
+        parser_map: defaultdict[str | FieldInfo, dict[int | None | str, _CliArg]],
         **values: Any,
     ) -> None:
         super().__init__(**values)
@@ -132,12 +130,12 @@ def __init__(
             parser_map[self.field_info][index] = parser_map[alias_path_dest][index]
 
     @classmethod
-    def get_kebab_case(cls, name: str, kebab_case: Optional[Union[bool, Literal['all', 'no_enums']]]) -> str:
+    def get_kebab_case(cls, name: str, kebab_case: bool | Literal['all', 'no_enums'] | None) -> str:
         return name.replace('_', '-') if kebab_case not in (None, False) else name
 
     @classmethod
     def get_enum_names(
-        cls, annotation: type[Any], kebab_case: Optional[Union[bool, Literal['all', 'no_enums']]]
+        cls, annotation: type[Any], kebab_case: bool | Literal['all', 'no_enums'] | None
     ) -> tuple[str, ...]:
         enum_names: tuple[str, ...] = ()
         annotation = _strip_annotated(annotation)
@@ -157,7 +155,7 @@ def field_info(self) -> FieldInfo:
         return self._field_info
 
     @cached_property
-    def subcommand_dest(self) -> Optional[str]:
+    def subcommand_dest(self) -> str | None:
         return f'{self.arg_prefix}:subcommand' if _CliSubCommand in self.field_info.metadata else None
 
     @cached_property
@@ -206,7 +204,7 @@ def alias_names(self) -> tuple[str, ...]:
         return self._alias_names
 
     @cached_property
-    def alias_paths(self) -> dict[str, Optional[int]]:
+    def alias_paths(self) -> dict[str, int | None]:
         return self._alias_paths
 
     @cached_property
@@ -236,7 +234,7 @@ def is_no_decode(self) -> bool:
 
 
 T = TypeVar('T')
-CliSubCommand = Annotated[Union[T, None], _CliSubCommand]
+CliSubCommand = Annotated[T | None, _CliSubCommand]
 CliPositionalArg = Annotated[T, _CliPositionalArg]
 _CliBoolFlag = TypeVar('_CliBoolFlag', bound=bool)
 CliImplicitFlag = Annotated[_CliBoolFlag, _CliImplicitFlag]
@@ -585,9 +583,7 @@ def _is_nested_alias_path_only_workaround(
             return True
         return False
 
-    def _get_merge_parsed_list_types(
-        self, parsed_list: list[str], field_name: str
-    ) -> tuple[Optional[type], Optional[type]]:
+    def _get_merge_parsed_list_types(self, parsed_list: list[str], field_name: str) -> tuple[type | None, type | None]:
         merge_type = self._cli_dict_args.get(field_name, list)
         if (
             merge_type is list
@@ -606,7 +602,7 @@ def _get_merge_parsed_list_types(
 
     def _merged_list_to_str(self, merged_list: list[str], field_name: str) -> str:
         decode_list: list[str] = []
-        is_use_decode: Optional[bool] = None
+        is_use_decode: bool | None = None
         cli_arg_map = self._parser_map.get(field_name, {})
         for index, item in enumerate(merged_list):
             cli_arg = cli_arg_map.get(index)
@@ -867,7 +863,7 @@ def _parse_known_args(*args: Any, **kwargs: Any) -> Namespace:
         self._add_subparsers = self._connect_parser_method(add_subparsers_method, 'add_subparsers_method')
         self._formatter_class = formatter_class
         self._cli_dict_args: dict[str, type[Any] | None] = {}
-        self._parser_map: defaultdict[str | FieldInfo, dict[Optional[int] | str, _CliArg]] = defaultdict(dict)
+        self._parser_map: defaultdict[str | FieldInfo, dict[int | None | str, _CliArg]] = defaultdict(dict)
         self._add_parser_args(
             parser=self.root_parser,
             model=self.settings_cls,
@@ -892,7 +888,7 @@ def _add_parser_args(
         is_model_suppressed: bool = False,
     ) -> ArgumentParser:
         subparsers: Any = None
-        alias_path_args: dict[str, Optional[int]] = {}
+        alias_path_args: dict[str, int | None] = {}
         # Ignore model default if the default is a model and not a subclass of the current model.
         model_default = (
             None
@@ -1159,7 +1155,7 @@ def _add_parser_submodels(
     def _add_parser_alias_paths(
         self,
         parser: Any,
-        alias_path_args: dict[str, Optional[int]],
+        alias_path_args: dict[str, int | None],
         added_args: list[str],
         arg_prefix: str,
         subcommand_prefix: str,

pydantic_settings/sources/providers/env.py

@@ -5,13 +5,14 @@
 from typing import (
     TYPE_CHECKING,
     Any,
+    get_args,
+    get_origin,
 )
 
 from pydantic import Json, TypeAdapter, ValidationError
 from pydantic._internal._utils import deep_update, is_model_class
 from pydantic.dataclasses import is_pydantic_dataclass
 from pydantic.fields import FieldInfo
-from typing_extensions import get_args, get_origin
 from typing_inspection.introspection import is_union_origin
 
 from ...utils import _lenient_issubclass

pydantic_settings/sources/providers/gcp.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Iterator, Mapping
 from functools import cached_property
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from .env import EnvSettingsSource
 
@@ -33,7 +33,7 @@ def import_gcp_secret_manager() -> None:
         ) from e
 
 
-class GoogleSecretManagerMapping(Mapping[str, Optional[str]]):
+class GoogleSecretManagerMapping(Mapping[str, str | None]):
     _loaded_secrets: dict[str, str | None]
     _secret_client: SecretManagerServiceClient
 
@@ -140,7 +140,7 @@ def __init__(
             env_parse_enums=env_parse_enums,
         )
 
-    def _load_env_vars(self) -> Mapping[str, Optional[str]]:
+    def _load_env_vars(self) -> Mapping[str, str | None]:
         return GoogleSecretManagerMapping(
             self._secret_client, project_id=self._project_id, case_sensitive=self.case_sensitive
         )