Add support for AWS Secrets Manager

docs/index.md

@@ -1775,6 +1775,61 @@ Last, run your application inside a Docker container and supply your newly creat
 docker service create --name pydantic-with-secrets --secret my_secret_data pydantic-app:latest
 ```
 
+## AWS Secrets Manager
+
+You must set one parameter:
+
+- `secret_id`: The AWS secret id
+
+You must have the same naming convention in the key value in secret as in the field name. For example, if the key in secret is named `SqlServerPassword`, the field name must be the same. You can use an alias too.
+
+In AWS Secrets Manager, nested models are supported with the `--` separator in the key name. For example, `SqlServer--Password`.
+
+Arrays (e.g. `MySecret--0`, `MySecret--1`) are not supported.
+
+```py
+import os
+
+from pydantic import BaseModel
+
+from pydantic_settings import (
+    AWSSecretsManagerSettingsSource,
+    BaseSettings,
+    PydanticBaseSettingsSource,
+)
+
+
+class SubModel(BaseModel):
+    a: str
+
+
+class AWSSecretsManagerSettings(BaseSettings):
+    foo: str
+    bar: int
+    sub: SubModel
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        aws_secrets_manager_settings = AWSSecretsManagerSettingsSource(
+            settings_cls,
+            os.environ['AWS_SECRETS_MANAGER_SECRET_ID'],
+        )
+        return (
+            init_settings,
+            env_settings,
+            dotenv_settings,
+            file_secret_settings,
+            aws_secrets_manager_settings,
+        )
+```
+
 ## Azure Key Vault
 
 You must set two parameters:

pydantic_settings/__init__.py

@@ -2,6 +2,7 @@
 from .main import BaseSettings, CliApp, SettingsConfigDict
 from .sources import (
     CLI_SUPPRESS,
+    AWSSecretsManagerSettingsSource,
     AzureKeyVaultSettingsSource,
     CliExplicitFlag,
     CliImplicitFlag,
@@ -27,6 +28,7 @@
 
 __all__ = (
     'CLI_SUPPRESS',
+    'AWSSecretsManagerSettingsSource',
     'AzureKeyVaultSettingsSource',
     'BaseSettings',
     'CliApp',

pydantic_settings/sources/__init__.py

@@ -7,6 +7,7 @@
     PydanticBaseSettingsSource,
     get_subcommand,
 )
+from .providers.aws import AWSSecretsManagerSettingsSource
 from .providers.azure import AzureKeyVaultSettingsSource
 from .providers.cli import (
     CLI_SUPPRESS,
@@ -30,6 +31,7 @@
 __all__ = [
     'CLI_SUPPRESS',
     'ENV_FILE_SENTINEL',
+    'AWSSecretsManagerSettingsSource',
     'AzureKeyVaultSettingsSource',
     'CliExplicitFlag',
     'CliImplicitFlag',

pydantic_settings/sources/providers/__init__.py

@@ -1,5 +1,6 @@
 """Package containing individual source implementations."""
 
+from .aws import AWSSecretsManagerSettingsSource
 from .azure import AzureKeyVaultSettingsSource
 from .cli import (
     CliExplicitFlag,
@@ -19,6 +20,7 @@
 from .yaml import YamlConfigSettingsSource
 
 __all__ = [
+    'AWSSecretsManagerSettingsSource',
     'AzureKeyVaultSettingsSource',
     'CliExplicitFlag',
     'CliImplicitFlag',

pydantic_settings/sources/providers/aws.py

@@ -0,0 +1,69 @@
+from __future__ import annotations as _annotations  # important for BaseSettings import to work
+
+import json
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Optional
+
+from .env import EnvSettingsSource
+
+if TYPE_CHECKING:
+    from pydantic_settings.main import BaseSettings
+
+
+boto3_client = None
+SecretsManagerClient = None
+
+
+def import_aws_secrets_manager() -> None:
+    global boto3_client
+    global SecretsManagerClient
+
+    try:
+        from boto3 import client as boto3_client
+        from mypy_boto3_secretsmanager.client import SecretsManagerClient
+    except ImportError as e:
+        raise ImportError(
+            'AWS Secrets Manager dependencies are not installed, run `pip install pydantic-settings[aws-secrets-manager]`'
+        ) from e
+
+
+class AWSSecretsManagerSettingsSource(EnvSettingsSource):
+    _secret_id: str
+    _secretsmanager_client: SecretsManagerClient  # type: ignore
+
+    def __init__(
+        self,
+        settings_cls: type[BaseSettings],
+        secret_id: str,
+        env_prefix: str | None = None,
+        env_parse_none_str: str | None = None,
+        env_parse_enums: bool | None = None,
+    ) -> None:
+        import_aws_secrets_manager()
+        self._secretsmanager_client = boto3_client('secretsmanager')  # type: ignore
+        self._secret_id = secret_id
+        super().__init__(
+            settings_cls,
+            case_sensitive=True,
+            env_prefix=env_prefix,
+            env_nested_delimiter='--',
+            env_ignore_empty=False,
+            env_parse_none_str=env_parse_none_str,
+            env_parse_enums=env_parse_enums,
+        )
+
+    def _load_env_vars(self) -> Mapping[str, Optional[str]]:
+        response = self._secretsmanager_client.get_secret_value(SecretId=self._secret_id)  # type: ignore
+
+        return json.loads(response['SecretString'])
+
+    def __repr__(self) -> str:
+        return (
+            f'{self.__class__.__name__}(secret_id={self._secret_id!r}, '
+            f'env_nested_delimiter={self.env_nested_delimiter!r})'
+        )
+
+
+__all__ = [
+    'AWSSecretsManagerSettingsSource',
+]

pyproject.toml

@@ -50,6 +50,7 @@ dynamic = ['version']
 yaml = ["pyyaml>=6.0.1"]
 toml = ["tomli>=2.0.1"]
 azure-key-vault = ["azure-keyvault-secrets>=4.8.0", "azure-identity>=1.16.0"]
+aws-secrets-manager = ["boto3>=1.35.0", "boto3-stubs[secretsmanager]"]
 
 [project.urls]
 Homepage = 'https://github.com/pydantic/pydantic-settings'
@@ -66,20 +67,23 @@ linting = [
     "pyyaml",
     "ruff",
     "types-pyyaml",
+    "boto3-stubs[secretsmanager]",
 ]
 testing = [
     "coverage[toml]",
     "pytest",
     "pytest-examples",
     "pytest-mock",
     "pytest-pretty",
+    "moto[secretsmanager]",
 ]
 
 [tool.pytest.ini_options]
 testpaths = 'tests'
 filterwarnings = [
     'error',
     'ignore:This is a placeholder until pydantic-settings.*:UserWarning',
+    'ignore::DeprecationWarning:botocore.*:',
 ]
 
 [tool.coverage.run]

tests/test_source_aws_secrets_manager.py

@@ -0,0 +1,120 @@
+"""
+Test pydantic_settings.AWSSecretsManagerSettingsSource.
+"""
+
+import json
+import os
+
+import pytest
+
+try:
+    import yaml
+    from moto import mock_aws
+except ImportError:
+    yaml = None
+    mock_aws = None
+
+from pydantic import BaseModel, Field
+
+from pydantic_settings import (
+    AWSSecretsManagerSettingsSource,
+    BaseSettings,
+    PydanticBaseSettingsSource,
+)
+from pydantic_settings.sources.providers.aws import import_aws_secrets_manager
+
+try:
+    aws_secrets_manager = True
+    import_aws_secrets_manager()
+    import boto3
+
+    os.environ['AWS_DEFAULT_REGION'] = os.environ.get('AWS_DEFAULT_REGION', 'us-east-1')
+except ImportError:
+    aws_secrets_manager = False
+
+
+MODULE = 'pydantic_settings.sources'
+
+if not yaml:
+    pytest.skip('PyYAML is not installed', allow_module_level=True)
+
+
+@pytest.mark.skipif(not aws_secrets_manager, reason='pydantic-settings[aws-secrets-manager] is not installed')
+class TestAWSSecretsManagerSettingsSource:
+    """Test AWSSecretsManagerSettingsSource."""
+
+    @mock_aws
+    def test___init__(self) -> None:
+        """Test __init__."""
+
+        class AWSSecretsManagerSettings(BaseSettings):
+            """AWSSecretsManager settings."""
+
+        client = boto3.client('secretsmanager')
+        client.create_secret(Name='test-secret', SecretString='{}')
+
+        AWSSecretsManagerSettingsSource(AWSSecretsManagerSettings, 'test-secret')
+
+    @mock_aws
+    def test___call__(self) -> None:
+        """Test __call__."""
+
+        class SqlServer(BaseModel):
+            password: str = Field(..., alias='Password')
+
+        class AWSSecretsManagerSettings(BaseSettings):
+            """AWSSecretsManager settings."""
+
+            SqlServerUser: str
+            sql_server_user: str = Field(..., alias='SqlServerUser')
+            sql_server: SqlServer = Field(..., alias='SqlServer')
+
+        expected_secret_value = 'SecretValue'
+        secret_data = {'SqlServerUser': expected_secret_value, 'SqlServer--Password': expected_secret_value}
+
+        client = boto3.client('secretsmanager')
+        client.create_secret(Name='test-secret', SecretString=json.dumps(secret_data))
+
+        obj = AWSSecretsManagerSettingsSource(AWSSecretsManagerSettings, 'test-secret')
+
+        settings = obj()
+
+        assert settings['SqlServerUser'] == expected_secret_value
+        assert settings['SqlServer']['Password'] == expected_secret_value
+
+    @mock_aws
+    def test_aws_secrets_manager_settings_source(self) -> None:
+        """Test AWSSecretsManagerSettingsSource."""
+
+        class SqlServer(BaseModel):
+            password: str = Field(..., alias='Password')
+
+        class AWSSecretsManagerSettings(BaseSettings):
+            """AWSSecretsManager settings."""
+
+            SqlServerUser: str
+            sql_server_user: str = Field(..., alias='SqlServerUser')
+            sql_server: SqlServer = Field(..., alias='SqlServer')
+
+            @classmethod
+            def settings_customise_sources(
+                cls,
+                settings_cls: type[BaseSettings],
+                init_settings: PydanticBaseSettingsSource,
+                env_settings: PydanticBaseSettingsSource,
+                dotenv_settings: PydanticBaseSettingsSource,
+                file_secret_settings: PydanticBaseSettingsSource,
+            ) -> tuple[PydanticBaseSettingsSource, ...]:
+                return (AWSSecretsManagerSettingsSource(settings_cls, 'test-secret'),)
+
+        expected_secret_value = 'SecretValue'
+        secret_data = {'SqlServerUser': expected_secret_value, 'SqlServer--Password': expected_secret_value}
+
+        client = boto3.client('secretsmanager')
+        client.create_secret(Name='test-secret', SecretString=json.dumps(secret_data))
+
+        settings = AWSSecretsManagerSettings()  # type: ignore
+
+        assert settings.SqlServerUser == expected_secret_value
+        assert settings.sql_server_user == expected_secret_value
+        assert settings.sql_server.password == expected_secret_value