Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions pydantic_settings/sources/providers/aws.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
from __future__ import annotations as _annotations # important for BaseSettings import to work
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 ..utils import parse_env_vars
from .env import EnvSettingsSource

if TYPE_CHECKING:
Expand Down Expand Up @@ -36,6 +39,7 @@ def __init__(
settings_cls: type[BaseSettings],
secret_id: str,
region_name: str | None = None,
case_sensitive: bool | None = True,
env_prefix: str | None = None,
env_parse_none_str: str | None = None,
env_parse_enums: bool | None = None,
Expand All @@ -45,7 +49,7 @@ def __init__(
self._secret_id = secret_id
super().__init__(
settings_cls,
case_sensitive=True,
case_sensitive=case_sensitive,
env_prefix=env_prefix,
env_nested_delimiter='--',
env_ignore_empty=False,
Expand All @@ -56,7 +60,12 @@ def __init__(
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'])
return parse_env_vars(
json.loads(response['SecretString']),
self.case_sensitive,
self.env_ignore_empty,
self.env_parse_none_str,
)

def __repr__(self) -> str:
return (
Expand Down
52 changes: 49 additions & 3 deletions tests/test_source_aws_secrets_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,10 @@
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')
@pytest.mark.skipif(
not aws_secrets_manager,
reason='pydantic-settings[aws-secrets-manager] is not installed',
)
class TestAWSSecretsManagerSettingsSource:
"""Test AWSSecretsManagerSettingsSource."""

Expand Down Expand Up @@ -76,7 +79,10 @@ class AWSSecretsManagerSettings(BaseSettings):
sql_server_user: str = Field(..., alias='SqlServerUser')
sql_server: SqlServer = Field(..., alias='SqlServer')

secret_data = {'SqlServerUser': 'test-user', 'SqlServer--Password': 'test-password'}
secret_data = {
'SqlServerUser': 'test-user',
'SqlServer--Password': 'test-password',
}

client = boto3.client('secretsmanager')
client.create_secret(Name='test-secret', SecretString=json.dumps(secret_data))
Expand All @@ -88,6 +94,43 @@ class AWSSecretsManagerSettings(BaseSettings):
assert settings['SqlServerUser'] == 'test-user'
assert settings['SqlServer']['Password'] == 'test-password'

@mock_aws
def test_secret_manager_case_insensitive_success(self) -> None:
"""Test secret manager getitem case insensitive success."""

class SqlServer(BaseModel):
password: str = Field(..., alias='Password')

class AWSSecretsManagerSettings(BaseSettings):
"""AWSSecretsManager settings."""

sql_server_user: str
sql_server: 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', case_sensitive=False),)

secret_data = {
'SQL_SERVER_USER': 'test-user',
'SQL_SERVER--PASSWORD': 'test-password',
}

client = boto3.client('secretsmanager')
client.create_secret(Name='test-secret', SecretString=json.dumps(secret_data))

settings = AWSSecretsManagerSettings() # type: ignore

assert settings.sql_server_user == 'test-user'
assert settings.sql_server.password == 'test-password'

@mock_aws
def test_aws_secrets_manager_settings_source(self) -> None:
"""Test AWSSecretsManagerSettingsSource."""
Expand All @@ -112,7 +155,10 @@ def settings_customise_sources(
) -> tuple[PydanticBaseSettingsSource, ...]:
return (AWSSecretsManagerSettingsSource(settings_cls, 'test-secret'),)

secret_data = {'SqlServerUser': 'test-user', 'SqlServer--Password': 'test-password'}
secret_data = {
'SqlServerUser': 'test-user',
'SqlServer--Password': 'test-password',
}

client = boto3.client('secretsmanager')
client.create_secret(Name='test-secret', SecretString=json.dumps(secret_data))
Expand Down