docs/index.md

@@ -156,7 +156,8 @@ class Settings(BaseSettings):
 ```
 
 !!! note
-    The default `env_prefix` is `''` (empty string).
+    The default `env_prefix` is `''` (empty string). `env_prefix` is not only for env settings but also for
+    dotenv files, secrets, and other sources.
 
 If you want to change the environment variable name for a single field, you can use an alias.
 
@@ -323,6 +324,58 @@ print(Settings().model_dump())
 `env_nested_delimiter` can be configured via the `model_config` as shown above, or via the
 `_env_nested_delimiter` keyword argument on instantiation.
 
+By default environment variables are split by `env_nested_delimiter` into arbitrarily deep nested fields. You can limit
+the depth of the nested fields with the `env_nested_max_split` config setting. A common use case this is particularly useful
+is for two-level deep settings, where the `env_nested_delimiter` (usually a single `_`) may be a substring of model
+field names. For example:
+
+```bash
+# your environment
+export GENERATION_LLM_PROVIDER='anthropic'
+export GENERATION_LLM_API_KEY='your-api-key'
+export GENERATION_LLM_API_VERSION='2024-03-15'
+```
+
+You could load them into the following settings model:
+
+```py
+from pydantic import BaseModel
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class LLMConfig(BaseModel):
+    provider: str = 'openai'
+    api_key: str
+    api_type: str = 'azure'
+    api_version: str = '2023-03-15-preview'
+
+
+class GenerationConfig(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_nested_delimiter='_', env_nested_max_split=1, env_prefix='GENERATION_'
+    )
+
+    llm: LLMConfig
+    ...
+
+
+print(GenerationConfig().model_dump())
+"""
+{
+    'llm': {
+        'provider': 'anthropic',
+        'api_key': 'your-api-key',
+        'api_type': 'azure',
+        'api_version': '2024-03-15',
+    }
+}
+"""
+```
+
+Without `env_nested_max_split=1` set, `GENERATION_LLM_API_KEY` would be parsed as `llm.api.key` instead of `llm.api_key`
+and it would raise a `ValidationError`.
+
 Nested environment variables take precedence over the top-level environment variable JSON
 (e.g. in the example above, `SUB_MODEL__V2` trumps `SUB_MODEL`).
 
@@ -842,9 +895,10 @@ print(User().model_dump())
 
 ### Subcommands and Positional Arguments
 
-Subcommands and positional arguments are expressed using the `CliSubCommand` and `CliPositionalArg` annotations. These
-annotations can only be applied to required fields (i.e. fields that do not have a default value). Furthermore,
-subcommands must be a valid type derived from either a pydantic `BaseModel` or pydantic.dataclasses `dataclass`.
+Subcommands and positional arguments are expressed using the `CliSubCommand` and `CliPositionalArg` annotations. The
+subcommand annotation can only be applied to required fields (i.e. fields that do not have a default value).
+Furthermore, subcommands must be a valid type derived from either a pydantic `BaseModel` or pydantic.dataclasses
+`dataclass`.
 
 Parsed subcommands can be retrieved from model instances using the `get_subcommand` utility function. If a subcommand is
 not required, set the `is_required` flag to `False` to disable raising an error if no subcommand is found.
@@ -1059,6 +1113,72 @@ For `BaseModel` and `pydantic.dataclasses.dataclass` types, `CliApp.run` will in
 * `cli_implicit_flags=True`
 * `cli_kebab_case=True`
 
+### Asynchronous CLI Commands
+
+Pydantic settings supports running asynchronous CLI commands via `CliApp.run` and `CliApp.run_subcommand`. With this feature, you can define async def methods within your Pydantic models (including subcommands) and have them executed just like their synchronous counterparts. Specifically:
+
+1. Asynchronous methods are supported: You can now mark your cli_cmd or similar CLI entrypoint methods as async def and have CliApp execute them.
+2. Subcommands may also be asynchronous: If you have nested CLI subcommands, the final (lowest-level) subcommand methods can likewise be asynchronous.
+3. Limit asynchronous methods to final subcommands: Defining parent commands as asynchronous is not recommended, because it can result in additional threads and event loops being created. For best performance and to avoid unnecessary resource usage, only implement your deepest (child) subcommands as async def.
+
+Below is a simple example demonstrating an asynchronous top-level command:
+
+```py
+from pydantic_settings import BaseSettings, CliApp
+
+
+class AsyncSettings(BaseSettings):
+    async def cli_cmd(self) -> None:
+        print('Hello from an async CLI method!')
+        #> Hello from an async CLI method!
+
+
+# If an event loop is already running, a new thread will be used;
+# otherwise, asyncio.run() is used to execute this async method.
+assert CliApp.run(AsyncSettings, cli_args=[]).model_dump() == {}
+```
+
+#### Asynchronous Subcommands
+
+As mentioned above, you can also define subcommands as async. However, only do so for the leaf (lowest-level) subcommand to avoid spawning new threads and event loops unnecessarily in parent commands:
+
+```py
+from pydantic import BaseModel
+
+from pydantic_settings import (
+    BaseSettings,
+    CliApp,
+    CliPositionalArg,
+    CliSubCommand,
+)
+
+
+class Clone(BaseModel):
+    repository: CliPositionalArg[str]
+    directory: CliPositionalArg[str]
+
+    async def cli_cmd(self) -> None:
+        # Perform async tasks here, e.g. network or I/O operations
+        print(f'Cloning async from "{self.repository}" into "{self.directory}"')
+        #> Cloning async from "repo" into "dir"
+
+
+class Git(BaseSettings):
+    clone: CliSubCommand[Clone]
+
+    def cli_cmd(self) -> None:
+        # Run the final subcommand (clone/init). It is recommended to define async methods only at the deepest level.
+        CliApp.run_subcommand(self)
+
+
+CliApp.run(Git, cli_args=['clone', 'repo', 'dir']).model_dump() == {
+    'repository': 'repo',
+    'directory': 'dir',
+}
+```
+
+When executing a subcommand with an asynchronous cli_cmd, Pydantic settings automatically detects whether the current thread already has an active event loop. If so, the async command is run in a fresh thread to avoid conflicts. Otherwise, it uses asyncio.run() in the current thread. This handling ensures your asynchronous subcommands “just work” without additional manual setup.
+
 ### Mutually Exclusive Groups
 
 CLI mutually exclusive groups can be created by inheriting from the `CliMutuallyExclusiveGroup` class.
@@ -1284,6 +1404,9 @@ However, if your use case [aligns more with #2](#command-line-support), using Py
 likely want required fields to be _strictly required at the CLI_. We can enable this behavior by using
 `cli_enforce_required`.
 
+!!! note
+    A required `CliPositionalArg` field is always strictly required (enforced) at the CLI.
+
 ```py
 import os
 import sys

pydantic_settings/main.py

@@ -1,5 +1,8 @@
 from __future__ import annotations as _annotations
 
+import asyncio
+import inspect
+import threading
 from argparse import Namespace
 from types import SimpleNamespace
 from typing import Any, ClassVar, TypeVar
@@ -38,6 +41,7 @@ class SettingsConfigDict(ConfigDict, total=False):
     env_file_encoding: str | None
     env_ignore_empty: bool
     env_nested_delimiter: str | None
+    env_nested_max_split: int | None
     env_parse_none_str: str | None
     env_parse_enums: bool | None
     cli_prog_name: str | None
@@ -112,6 +116,7 @@ class BaseSettings(BaseModel):
         _env_file_encoding: The env file encoding, e.g. `'latin-1'`. Defaults to `None`.
         _env_ignore_empty: Ignore environment variables where the value is an empty string. Default to `False`.
         _env_nested_delimiter: The nested env values delimiter. Defaults to `None`.
+        _env_nested_max_split: The nested env values maximum nesting. Defaults to `None`, which means no limit.
         _env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.)
             into `None` type(None). Defaults to `None` type(None), which means no parsing should occur.
         _env_parse_enums: Parse enum field names to values. Defaults to `None.`, which means no parsing should occur.
@@ -140,19 +145,21 @@ class BaseSettings(BaseModel):
     """
 
     def __init__(
-        __pydantic_self__,
+        self,
+        /,
         _case_sensitive: bool | None = None,
         _nested_model_default_partial_update: bool | None = None,
         _env_prefix: str | None = None,
         _env_file: DotenvType | None = ENV_FILE_SENTINEL,
         _env_file_encoding: str | None = None,
         _env_ignore_empty: bool | None = None,
         _env_nested_delimiter: str | None = None,
+        _env_nested_max_split: int | None = None,
         _env_parse_none_str: str | None = None,
         _env_parse_enums: bool | None = None,
         _cli_prog_name: str | None = None,
         _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,
-        _cli_settings_source: CliSettingsSource[Any] | None = None,
+        _cli_settings_source: CliSettingsSource[object] | None = None,
         _cli_parse_none_str: str | None = None,
         _cli_hide_none_type: bool | None = None,
         _cli_avoid_json: bool | None = None,
@@ -167,9 +174,8 @@ def __init__(
         _secrets_dir: PathType | None = None,
         **values: Any,
     ) -> None:
-        # Uses something other than `self` the first arg to allow "self" as a settable attribute
         super().__init__(
-            **__pydantic_self__._settings_build_values(
+            **self._settings_build_values(
                 values,
                 _case_sensitive=_case_sensitive,
                 _nested_model_default_partial_update=_nested_model_default_partial_update,
@@ -178,6 +184,7 @@ def __init__(
                 _env_file_encoding=_env_file_encoding,
                 _env_ignore_empty=_env_ignore_empty,
                 _env_nested_delimiter=_env_nested_delimiter,
+                _env_nested_max_split=_env_nested_max_split,
                 _env_parse_none_str=_env_parse_none_str,
                 _env_parse_enums=_env_parse_enums,
                 _cli_prog_name=_cli_prog_name,
@@ -232,6 +239,7 @@ def _settings_build_values(
         _env_file_encoding: str | None = None,
         _env_ignore_empty: bool | None = None,
         _env_nested_delimiter: str | None = None,
+        _env_nested_max_split: int | None = None,
         _env_parse_none_str: str | None = None,
         _env_parse_enums: bool | None = None,
         _cli_prog_name: str | None = None,
@@ -270,6 +278,11 @@ def _settings_build_values(
             if _env_nested_delimiter is not None
             else self.model_config.get('env_nested_delimiter')
         )
+        env_nested_max_split = (
+            _env_nested_max_split
+            if _env_nested_max_split is not None
+            else self.model_config.get('env_nested_max_split')
+        )
         env_parse_none_str = (
             _env_parse_none_str if _env_parse_none_str is not None else self.model_config.get('env_parse_none_str')
         )
@@ -333,6 +346,7 @@ def _settings_build_values(
             case_sensitive=case_sensitive,
             env_prefix=env_prefix,
             env_nested_delimiter=env_nested_delimiter,
+            env_nested_max_split=env_nested_max_split,
             env_ignore_empty=env_ignore_empty,
             env_parse_none_str=env_parse_none_str,
             env_parse_enums=env_parse_enums,
@@ -344,6 +358,7 @@ def _settings_build_values(
             case_sensitive=case_sensitive,
             env_prefix=env_prefix,
             env_nested_delimiter=env_nested_delimiter,
+            env_nested_max_split=env_nested_max_split,
             env_ignore_empty=env_ignore_empty,
             env_parse_none_str=env_parse_none_str,
             env_parse_enums=env_parse_enums,
@@ -412,6 +427,7 @@ def _settings_build_values(
         env_file_encoding=None,
         env_ignore_empty=False,
         env_nested_delimiter=None,
+        env_nested_max_split=None,
         env_parse_none_str=None,
         env_parse_enums=None,
         cli_prog_name=None,
@@ -446,10 +462,48 @@ class CliApp:
 
     @staticmethod
     def _run_cli_cmd(model: Any, cli_cmd_method_name: str, is_required: bool) -> Any:
-        if hasattr(type(model), cli_cmd_method_name):
-            getattr(type(model), cli_cmd_method_name)(model)
-        elif is_required:
-            raise SettingsError(f'Error: {type(model).__name__} class is missing {cli_cmd_method_name} entrypoint')
+        command = getattr(type(model), cli_cmd_method_name, None)
+        if command is None:
+            if is_required:
+                raise SettingsError(f'Error: {type(model).__name__} class is missing {cli_cmd_method_name} entrypoint')
+            return model
+
+        # If the method is asynchronous, we handle its execution based on the current event loop status.
+        if inspect.iscoroutinefunction(command):
+            # For asynchronous methods, we have two execution scenarios:
+            # 1. If no event loop is running in the current thread, run the coroutine directly with asyncio.run().
+            # 2. If an event loop is already running in the current thread, run the coroutine in a separate thread to avoid conflicts.
+            try:
+                # Check if an event loop is currently running in this thread.
+                loop = asyncio.get_running_loop()
+            except RuntimeError:
+                loop = None
+
+            if loop and loop.is_running():
+                # We're in a context with an active event loop (e.g., Jupyter Notebook).
+                # Running asyncio.run() here would cause conflicts, so we use a separate thread.
+                exception_container = []
+
+                def run_coro() -> None:
+                    try:
+                        # Execute the coroutine in a new event loop in this separate thread.
+                        asyncio.run(command(model))
+                    except Exception as e:
+                        exception_container.append(e)
+
+                thread = threading.Thread(target=run_coro)
+                thread.start()
+                thread.join()
+                if exception_container:
+                    # Propagate exceptions from the separate thread.
+                    raise exception_container[0]
+            else:
+                # No event loop is running; safe to run the coroutine directly.
+                asyncio.run(command(model))
+        else:
+            # For synchronous methods, call them directly.
+            command(model)
+
         return model
 
     @staticmethod