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
66 lines (50 loc) · 2.47 KB
/
session.py
File metadata and controls
66 lines (50 loc) · 2.47 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
import os
import boto3.session
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)
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):
if service_name not in self._service_endpoint_mapping:
raise Exception('%s is not supported by this mock session.' % (service_name))
return boto3.client(service_name, endpoint_url=self._service_endpoint_mapping[service_name],
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 resource(self, service_name, **kwargs):
if service_name not in self._service_endpoint_mapping:
raise Exception('%s is not supported by this mock session.' % (service_name))
return boto3.resource(service_name,
endpoint_url=self._service_endpoint_mapping[service_name],
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_default_session():
global DEFAULT_SESSION
if DEFAULT_SESSION is None:
DEFAULT_SESSION = Session()
return DEFAULT_SESSION
def client(*args, **kwargs):
return _get_default_session().client(*args, **kwargs)
def resource(*args, **kwargs):
return _get_default_session().resource(*args, **kwargs)