forked from localstack/localstack-python-client
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconfig.py
More file actions
64 lines (54 loc) 路 2.11 KB
/
Copy pathconfig.py
File metadata and controls
64 lines (54 loc) 路 2.11 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
import os
import json
from six.moves.urllib.parse import urlparse
_service_endpoints_template = {
'apigateway': '{proto}://{host}:4567',
'kinesis': '{proto}://{host}:4568',
'dynamodb': '{proto}://{host}:4569',
'dynamodbstreams': '{proto}://{host}:4570',
'elasticsearch': '{proto}://{host}:4571',
's3': '{proto}://{host}:4572',
'firehose': '{proto}://{host}:4573',
'lambda': '{proto}://{host}:4574',
'sns': '{proto}://{host}:4575',
'sqs': '{proto}://{host}:4576',
'redshift': '{proto}://{host}:4577',
'es': '{proto}://{host}:4578',
'ses': '{proto}://{host}:4579',
'route53': '{proto}://{host}:4580',
'cloudformation': '{proto}://{host}:4581',
'cloudwatch': '{proto}://{host}:4582',
'ssm': '{proto}://{host}:4583',
'secretsmanager': '{proto}://{host}:4584',
'stepfunctions': '{proto}://{host}:4585',
'logs': '{proto}://{host}:4586',
'events': '{proto}://{host}:4587',
'elb': '{proto}://{host}:4588',
'iot': '{proto}://{host}:4589',
'cognito-idp': '{proto}://{host}:4590',
'cognito-identity': '{proto}://{host}:4591',
'sts': '{proto}://{host}:4592',
'iam': '{proto}://{host}:4593',
'rds': '{proto}://{host}:4594',
'cloudsearch': '{proto}://{host}:4595',
'swf': '{proto}://{host}:4596',
'ec2': '{proto}://{host}:4597'
}
def get_service_endpoint(service, localstack_host=None):
endpoints = get_service_endpoints(localstack_host=localstack_host)
return endpoints.get(service)
def get_service_endpoints(localstack_host=None):
if localstack_host is None:
localstack_host = os.environ.get('LOCALSTACK_HOST', 'localhost')
protocol = 'https' if os.environ.get('USE_SSL') in ('1', 'true') else 'http'
return json.loads(json.dumps(_service_endpoints_template)
.replace('{proto}', protocol).replace('{host}', localstack_host))
def get_service_port(service):
ports = get_service_ports()
return ports.get(service)
def get_service_ports():
endpoints = get_service_endpoints()
result = {}
for service, url in endpoints.items():
result[service] = urlparse(url).port
return result