Coverage for src/extratools_cloud/aws/secrets_manager.py: 0%

1import json 

2from collections.abc import Iterable 

3from typing import Literal 

4 

5from botocore.client import BaseClient 

6from extratools_core.crudl import CRUDLDict 

7from extratools_core.json import JsonDict 

8from extratools_core.str import decode, encode 

9 

10from .helpers import ClientErrorHandler, get_client 

11 

12default_client: BaseClient = get_client("secretsmanager") 

13 

14 

15def get_resource_dict( 

16 *, 

17 secret_name_prefix: str | None = None, 

18 client: BaseClient | None = None, 

19 encoding: Literal["gzip", "zstd"] | None = None, 

20 force_delete: bool = False, 

21) -> CRUDLDict[str, JsonDict]: 

22 client = client or default_client 

23 

24 # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html 

25 

26 def check_secret_name(secret_name: str) -> str: 

27 if secret_name_prefix and not secret_name.startswith(secret_name_prefix): 

28 raise ValueError 

29 

30 return secret_name 

31 

32 def create_func(secret_name: str | None, value: JsonDict) -> None: 

33 if secret_name is None: 

34 raise ValueError 

35 

36 client.create_secret( 

37 Name=check_secret_name(secret_name), 

38 # It is possible to use `SecretBinary` to store compressed binary directly 

39 # without further Base64 encoding to store compressed string with 33%+ size overhead. 

40 # https://en.wikipedia.org/wiki/Base64 

41 # However, it would also make code more complex. 

42 SecretString=encode(json.dumps(value), encoding=encoding), 

43 ) 

44 

45 @ClientErrorHandler( 

46 "ResourceNotFoundException", 

47 KeyError, 

48 ) 

49 def read_func(secret_name: str) -> JsonDict: 

50 return json.loads(decode( 

51 client.get_secret_value( 

52 SecretId=check_secret_name(secret_name), 

53 )["SecretString"], 

54 encoding=encoding, 

55 )) 

56 

57 def list_func(_: None) -> Iterable[tuple[str, None]]: 

58 paginator = client.get_paginator("list_secrets") 

59 for page in paginator.paginate( 

60 **({} if secret_name_prefix is None else dict( 

61 Filters=[{ 

62 "Key": "name", 

63 "Values": [secret_name_prefix], 

64 }], 

65 )), 

66 ): 

67 for secret in page.get("SecretList", []): 

68 # It seems LocalStack always return delete secret even if 

69 # `IncludePlannedDeletion` is set to false. 

70 if "DeletedDate" not in secret: 

71 yield (secret["Name"], None) 

72 

73 return CRUDLDict[str, JsonDict]( 

74 create_func=create_func, 

75 read_func=read_func, 

76 update_func=lambda secret_name, value: client.put_secret_value( 

77 SecretId=check_secret_name(secret_name), 

78 SecretString=encode(json.dumps(value), encoding=encoding), 

79 ), 

80 delete_func=lambda secret_name: client.delete_secret( 

81 SecretId=check_secret_name(secret_name), 

82 ForceDeleteWithoutRecovery=force_delete, 

83 ), 

84 list_func=list_func, 

85 )