-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
boto_utils.py
142 lines (111 loc) · 4.33 KB
/
boto_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
"""
This module contains utility functions for boto3 library
"""
from typing import Any, Optional
from boto3 import Session
from botocore.config import Config
from botocore.exceptions import ClientError
from typing_extensions import Protocol
from samcli import __version__
from samcli.cli.global_config import GlobalConfig
def get_boto_config_with_user_agent(**kwargs) -> Config:
"""
Automatically add user agent string to boto configs.
Parameters
----------
kwargs :
key=value params which will be added to the Config object
Returns
-------
Config
Returns config instance which contains given parameters in it
"""
gc = GlobalConfig()
return Config(
user_agent_extra=(
f"aws-sam-cli/{__version__}/{gc.installation_id}" if gc.telemetry_enabled else f"aws-sam-cli/{__version__}"
),
**kwargs,
)
# Type definition of following boto providers, which is equal to Callable[[str], Any]
class BotoProviderType(Protocol):
def __call__(self, service_name: str) -> Any: ... # pragma: no cover
def get_boto_client_provider_from_session_with_config(session: Session, **kwargs) -> BotoProviderType:
"""
Returns a wrapper function for boto client with given configuration. It can be used like;
client_provider = get_boto_client_wrapper_with_config(session=session)
lambda_client = client_provider("lambda")
Parameters
----------
session: Session
Boto3 session object
kwargs :
Key-value params that will be passed to get_boto_config_with_user_agent
Returns
-------
A callable function which will return a boto client
"""
return lambda client_name: session.client(client_name, config=get_boto_config_with_user_agent(**kwargs))
def get_boto_client_provider_with_config(
region: Optional[str] = None, profile: Optional[str] = None, **kwargs
) -> BotoProviderType:
"""
Returns a wrapper function for boto client with given configuration. It can be used like;
client_provider = get_boto_client_wrapper_with_config(region_name=region)
lambda_client = client_provider("lambda")
Parameters
----------
region: Optional[str]
AWS region name
profile: Optional[str]
Profile name from credentials
kwargs :
Key-value params that will be passed to get_boto_config_with_user_agent
Returns
-------
A callable function which will return a boto client
"""
return get_boto_client_provider_from_session_with_config(
Session(region_name=region, profile_name=profile), **kwargs
)
def get_boto_resource_provider_from_session_with_config(session: Session, **kwargs) -> BotoProviderType:
"""
Returns a wrapper function for boto resource with given configuration. It can be used like;
resource_provider = get_boto_resource_wrapper_with_config(session=session)
cloudformation_resource = resource_provider("cloudformation")
Parameters
----------
session: Session
Boto3 session object
kwargs :
Key-value params that will be passed to get_boto_config_with_user_agent
Returns
-------
A callable function which will return a boto resource
"""
return lambda resource_name: session.resource(resource_name, config=get_boto_config_with_user_agent(**kwargs))
def get_boto_resource_provider_with_config(
region: Optional[str] = None, profile: Optional[str] = None, **kwargs
) -> BotoProviderType:
"""
Returns a wrapper function for boto resource with given configuration. It can be used like;
resource_provider = get_boto_resource_wrapper_with_config(region_name=region)
cloudformation_resource = resource_provider("cloudformation")
Parameters
----------
region: Optional[str]
AWS region name
profile: Optional[str]
Profile name from credentials
kwargs :
Key-value params that will be passed to get_boto_config_with_user_agent
Returns
-------
A callable function which will return a boto resource
"""
return get_boto_resource_provider_from_session_with_config(
Session(region_name=region, profile_name=profile), **kwargs
)
def get_client_error_code(client_error: ClientError) -> Optional[str]:
"""Extracts error code from boto ClientError"""
return client_error.response.get("Error", {}).get("Code")