-
Notifications
You must be signed in to change notification settings - Fork 4.3k
/
Copy pathaddmodel.py
120 lines (98 loc) · 4.75 KB
/
addmodel.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
# Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import json
import os
from botocore.model import ServiceModel
from awscli.customizations.commands import BasicCommand
def _get_endpoint_prefix_to_name_mappings(session):
# Get the mappings of endpoint prefixes to service names from the
# available service models.
prefixes_to_services = {}
for service_name in session.get_available_services():
service_model = session.get_service_model(service_name)
prefixes_to_services[service_model.endpoint_prefix] = service_name
return prefixes_to_services
def _get_service_name(session, endpoint_prefix):
if endpoint_prefix in session.get_available_services():
# Check if the endpoint prefix is a pre-existing service.
# If it is, use that endpoint prefix as the service name.
return endpoint_prefix
else:
# The service may have a different endpoint prefix than its name
# So we need to determine what the correct mapping may be.
# Figure out the mappings of endpoint prefix to service names.
name_mappings = _get_endpoint_prefix_to_name_mappings(session)
# Determine the service name from the mapping.
# If it does not exist in the mapping, return the original endpoint
# prefix.
return name_mappings.get(endpoint_prefix, endpoint_prefix)
def get_model_location(session, service_definition, service_name=None):
"""Gets the path of where a service-2.json file should go in ~/.aws/models
:type session: botocore.session.Session
:param session: A session object
:type service_definition: dict
:param service_definition: The json loaded service definition
:type service_name: str
:param service_name: The service name to use. If this not provided,
this will be determined from a combination of available services
and the service definition.
:returns: The path to where are model should be placed based on
the service definition and the current services in botocore.
"""
# Add the ServiceModel abstraction over the service json definition to
# make it easier to work with.
service_model = ServiceModel(service_definition)
# Determine the service_name if not provided
if service_name is None:
endpoint_prefix = service_model.endpoint_prefix
service_name = _get_service_name(session, endpoint_prefix)
api_version = service_model.api_version
# For the model location we only want the custom data path (~/.aws/models
# not the one set by AWS_DATA_PATH)
data_path = session.get_component('data_loader').CUSTOMER_DATA_PATH
# Use the version of the model to determine the file's naming convention.
service_model_name = (
'service-%d.json' % int(
float(service_definition.get('version', '2.0'))))
return os.path.join(data_path, service_name, api_version,
service_model_name)
class AddModelCommand(BasicCommand):
NAME = 'add-model'
DESCRIPTION = (
'Adds a service JSON model to the appropriate location in '
'~/.aws/models. Once the model gets added, CLI commands and Boto3 '
'clients will be immediately available for the service JSON model '
'provided.'
)
ARG_TABLE = [
{'name': 'service-model', 'required': True, 'help_text': (
'The contents of the service JSON model.')},
{'name': 'service-name', 'help_text': (
'Overrides the default name used by the service JSON '
'model to generate CLI service commands and Boto3 clients.')}
]
def _run_main(self, parsed_args, parsed_globals):
service_definition = json.loads(parsed_args.service_model)
# Get the path to where the model should be written
model_location = get_model_location(
self._session, service_definition, parsed_args.service_name
)
# If the service_name/api_version directories do not exist,
# then create them.
model_directory = os.path.dirname(model_location)
if not os.path.exists(model_directory):
os.makedirs(model_directory)
# Write the model to the specified location
with open(model_location, 'wb') as f:
f.write(parsed_args.service_model.encode('utf-8'))
return 0