Add support for AWS Secrets Manager

Author: mavwolverine

.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']
 
     env:
       PYTHON: ${{ matrix.python }}

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 Key Vault, nested models are supported with the `--` separator. For example, `SqlServer--Password`.
+
+Key Vault 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

@@ -1,6 +1,7 @@
 from .main import BaseSettings, CliApp, SettingsConfigDict
 from .sources import (
     CLI_SUPPRESS,
+    AWSSecretsManagerSettingsSource,
     AzureKeyVaultSettingsSource,
     CliExplicitFlag,
     CliImplicitFlag,
@@ -49,6 +50,7 @@
     'SettingsError',
     'TomlConfigSettingsSource',
     'YamlConfigSettingsSource',
+    'AWSSecretsManagerSettingsSource',
     'AzureKeyVaultSettingsSource',
     'get_subcommand',
     '__version__',

pydantic_settings/sources.py

@@ -96,6 +96,19 @@ def import_toml() -> None:
         import tomllib
 
 
+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
+
+
 def import_azure_key_vault() -> None:
     global TokenCredential
     global SecretClient
@@ -2188,6 +2201,43 @@ def __repr__(self) -> str:
         return f'{self.__class__.__name__}(yaml_file={self.yaml_file_path})'
 
 
+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)
+
+        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})'
+        )
+
+
 class AzureKeyVaultMapping(Mapping[str, Optional[str]]):
     _loaded_secrets: dict[str, str | None]
     _secret_client: SecretClient  # type: ignore

pyproject.toml

@@ -20,7 +20,6 @@ classifiers = [
     'Programming Language :: Python',
     'Programming Language :: Python :: 3',
     'Programming Language :: Python :: 3 :: Only',
-    'Programming Language :: Python :: 3.9',
     'Programming Language :: Python :: 3.10',
     'Programming Language :: Python :: 3.11',
     'Programming Language :: Python :: 3.12',
@@ -38,7 +37,7 @@ classifiers = [
     'Topic :: Software Development :: Libraries :: Python Modules',
     'Topic :: Internet',
 ]
-requires-python = '>=3.9'
+requires-python = '>=3.10'
 dependencies = [
     'pydantic>=2.7.0',
     'python-dotenv>=0.21.0',
@@ -47,9 +46,10 @@ dependencies = [
 dynamic = ['version']
 
 [project.optional-dependencies]
-yaml = ["pyyaml>=6.0.1"]
+yaml = ["pyyaml>=6.0.2"]
 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 +66,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 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