forked from localstack/localstack-python-client
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsession.py
More file actions
93 lines (73 loc) · 3.39 KB
/
session.py
File metadata and controls
93 lines (73 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
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 they are contained already
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 they are contained already
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)