-
Notifications
You must be signed in to change notification settings - Fork 30
Expand file tree
/
Copy pathsession.py
More file actions
109 lines (88 loc) · 3.39 KB
/
session.py
File metadata and controls
109 lines (88 loc) · 3.39 KB
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
from boto3 import client as boto3_client
from boto3 import resource as boto3_resource
from botocore.credentials import Credentials
from localstack_client import config
DEFAULT_SESSION = None
class Session(object):
"""
This is a custom LocalStack session used to
emulate the boto3.session object.
"""
def __init__(
self,
aws_access_key_id="accesskey",
aws_secret_access_key="secretkey",
aws_session_token="token",
region_name="us-east-1",
botocore_session=None,
profile_name=None,
localstack_host=None,
):
self.env = "local"
self.aws_access_key_id = aws_access_key_id
self.aws_secret_access_key = aws_secret_access_key
self.aws_session_token = aws_session_token
self.region_name = region_name
self._service_endpoint_mapping = config.get_service_endpoints(localstack_host)
self.common_protected_kwargs = {
"aws_access_key_id": self.aws_access_key_id,
"aws_secret_access_key": self.aws_secret_access_key,
"region_name": self.region_name,
"verify": False,
}
def get_credentials(self):
"""
Returns botocore.credential.Credential object.
"""
return Credentials(
access_key=self.aws_access_key_id,
secret_key=self.aws_secret_access_key,
token=self.aws_session_token,
)
def client(self, service_name, **kwargs):
"""
Mock boto3 client
If **kwargs are provided they will passed through to boto3.client unless already contained
within protected_kwargs which are set with priority
Returns boto3.resources.factory.s3.ServiceClient object
"""
if service_name not in self._service_endpoint_mapping:
raise Exception(
"%s is not supported by this mock session." % (service_name)
)
protected_kwargs = {
**self.common_protected_kwargs,
"service_name": service_name,
"endpoint_url": self._service_endpoint_mapping[service_name],
}
return boto3_client(**{**kwargs, **protected_kwargs})
def resource(self, service_name, **kwargs):
"""
Mock boto3 resource
If **kwargs are provided they will passed through to boto3.client unless already contained
within overwrite_kwargs which are set with priority
Returns boto3.resources.factory.s3.ServiceResource object
"""
if service_name not in self._service_endpoint_mapping:
raise Exception(
"%s is not supported by this mock session." % (service_name)
)
protected_kwargs = {
**self.common_protected_kwargs,
"service_name": service_name,
"endpoint_url": self._service_endpoint_mapping[service_name],
}
return boto3_resource(**{**kwargs, **protected_kwargs})
def _get_default_session():
global DEFAULT_SESSION
if DEFAULT_SESSION is None:
DEFAULT_SESSION = Session()
return DEFAULT_SESSION
def client(*args, **kwargs):
if kwargs:
return Session(**kwargs).client(*args, **kwargs)
return _get_default_session().client(*args, **kwargs)
def resource(*args, **kwargs):
if kwargs:
return Session(**kwargs).resource(*args, **kwargs)
return _get_default_session().resource(*args, **kwargs)