2
2
from pathlib import Path
3
3
from typing import Any , Dict , Optional , Union
4
4
5
+ import boto3
5
6
from pydantic import BaseSettings
6
7
7
- from .ssm import lazy_parameter
8
-
9
8
logger = logging .getLogger (__name__ )
10
9
11
10
@@ -15,12 +14,16 @@ class AwsSsmSettingsSource:
15
14
def __init__ (self , ssm_prefix : Union [Path , str , None ]):
16
15
self .ssm_prefix : Union [Path , str , None ] = ssm_prefix
17
16
17
+ @property
18
+ def client (self ):
19
+ return boto3 .client ("ssm" )
20
+
18
21
def __call__ (self , settings : BaseSettings ) -> Dict [str , Any ]:
19
22
"""
20
23
Returns lazy SSM values for all settings.
21
24
"""
22
25
secrets : Dict [str , Optional [Any ]] = {}
23
-
26
+
24
27
if self .ssm_prefix is None :
25
28
return secrets
26
29
@@ -31,10 +34,11 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
31
34
32
35
logger .debug (f"Building SSM settings with prefix of { secrets_path = } " )
33
36
34
- for field in settings .__fields__ .values ():
35
- for env_name in field .field_info .extra ["env_names" ]:
36
- secrets [field .alias ] = lazy_parameter (path = (secrets_path / env_name ), field = field )
37
- return secrets
37
+ params = self .client .get_parameters_by_path (Path = str (secrets_path ), WithDecryption = True )['Parameters' ]
38
+
39
+ return {
40
+ str (Path (param ['Name' ]).relative_to (secrets_path )): param ['Value' ] for param in params
41
+ }
38
42
39
43
def __repr__ (self ) -> str :
40
44
return f"AwsSsmSettingsSource(ssm_prefix={ self .ssm_prefix !r} )"
0 commit comments