-
Notifications
You must be signed in to change notification settings - Fork 3
/
api_create.py
198 lines (161 loc) · 8.73 KB
/
api_create.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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
import glob
import json
import os
import api_utils
#
# Exit codes
# ============
# 0 - Exit without issue
# 1 - Connection issue
# 2 - An input is not of the expected value
#
# Start execution
#
# 1. Load values from environment variables
#
# URLs and credentials of the API Manager deployment
# env var: WSO2_APIM_APIMGT_URL
apimgt_url = os.getenv("WSO2_APIM_APIMGT_URL", None)
# env var: WSO2_APIM_GW_URL
gw_url = os.getenv("WSO2_APIM_GW_URL", None)
# env var: WSO2_APIM_APIMGT_USERNAME
apimgt_username = os.getenv("WSO2_APIM_APIMGT_USERNAME", None)
# env var: WSO2_APIM_APIMGT_PASSWD
apimgt_pwd = os.getenv("WSO2_APIM_APIMGT_PASSWD", None)
# ignore (or not) TLS errors
# env var: WSO2_APIM_VERIFY_SSL
verify_ssl = os.getenv("WSO2_APIM_VERIFY_SSL") in ["True", "true", "yes", "1"]
# backend URLs to use when creating APIs
# env var: WSO2_APIM_BE_PROD
backend_url_prod = os.getenv("WSO2_APIM_BE_PROD", None)
# env var: WSO2_APIM_BE_SNDBX
backend_url_sandbox = os.getenv("WSO2_APIM_BE_SNDBX", None)
# The status at which the api should be created. Can be values from
# https://docs.wso2.com/display/AM260/Key+Concepts#KeyConcepts-APIlifecycle
# env var: WSO2_APIM_API_STATUS
api_status = os.getenv("WSO2_APIM_API_STATUS", "CREATED")
# Owner to be specified in the DCR
api_owner = os.getenv("WSO2_APIM_APIMGT_OWNER", None)
#
# 2. Sanitize input
#
# Any empty or invalid value should result in an immediate exit.
#
if apimgt_url is None:
print "[ERROR] API Manager URL is empty. Please set environment variable WSO2_APIM_APIMGT_URL"
exit(2)
if gw_url is None:
print "[ERROR] Gateway URL is empty. Please set environment variable WSO2_APIM_GW_URL"
exit(2)
if apimgt_username is None:
print "[ERROR] API Manager Username is empty. Please set environment variable WSO2_APIM_APIMGT_USERNAME"
exit(2)
if apimgt_pwd is None:
print "[ERROR] API Manager Password is empty. Please set environment variable WSO2_APIM_APIMGT_PASSWD"
exit(2)
if verify_ssl is None:
print "[ERROR] Verify_SSL flag is empty. Please set environment variable WSO2_APIM_VERIFY_SSL"
exit(2)
if backend_url_prod is None:
print "[ERROR] Production Backend URL is empty. Please set environment variable WSO2_APIM_BE_PROD"
exit(2)
if backend_url_sandbox is None:
print "[ERROR] Sandbox Backend URL is empty. Please set environment variable WSO2_APIM_BE_SNDBX"
exit(2)
if api_status is None:
print "[ERROR] API Status is empty. Please set environment variable WSO2_APIM_API_STATUS"
exit(2)
if api_status not in ["CREATED", "PROTOTYPED", "PUBLISHED", "BLOCKED", "DEPRECATED", "RETIRED"]:
print "[ERROR] API Status is an invalid value. Please set environment variable WSO2_APIM_API_STATUS with values " \
"found in https://docs.wso2.com/display/AM260/Key+Concepts#KeyConcepts-APIlifecycle"
exit(2)
if api_owner is None:
print "[ERROR] API Owner is empty. Please set environment variable WSO2_APIM_APIMGT_OWNER"
exit(2)
# Run the following for the Python script execution
if __name__ == '__main__':
print "Obtaining access token..."
access_token = api_utils.get_access_token(apimgt_url, gw_url, apimgt_username, apimgt_pwd, api_owner, verify_ssl)
if access_token is None:
print "[ERROR] Error when obtaining access token. Aborting..."
exit(1)
# Iterate through all json files inside api_definitions
api_requests = {}
for api_request_file in glob.glob("api_definitions/*_apis.json"):
request_json_content = json.loads(open(api_request_file).read())
requesting_user = request_json_content["username"]
department = request_json_content["department"]
print "Parsing create requests for [user] %s [department] %s [requests] %s..." % (
requesting_user, department, len(request_json_content["apis"]))
# For each api request, it should be checked whether an API with the same name already exists.
# If an API with the same name exists, it should be checked if there is an existing version matching the
# version requested.
#
# If there is no matching API, an API create request is done.
# If a matching API is there, but no matching version, a version would be added
# If both a matching API with a matching version exists, the API would not be created.
for api_request in request_json_content["apis"]:
api_name_exists, api_id = api_utils.api_name_exists(api_request["name"], apimgt_url, access_token,
verify_ssl)
if api_name_exists:
# Check if the same version exists
api_version_exists, _ = api_utils.api_version_exists(api_request["name"], api_request["version"],
apimgt_url, access_token, verify_ssl)
if api_version_exists:
# Do not create the same version
print "\t%s:%s \t :Exists. Skipping..." % (api_request["name"], api_request["version"])
continue
else:
# Add as a new version
print "\t%s:%s \t :Adding version..." % (api_request["name"], api_request["version"])
addVersionSuccessful, apiId = api_utils.add_api_version(api_id, api_request["version"], apimgt_url, access_token, verify_ssl)
if not addVersionSuccessful:
print "[ERROR] API new version addition failed. [server] %s. Continuing..." % apimgt_url
# Get apiId from create response and then publish
publishSuccessful = api_utils.change_lifecycle(apiId, "Publish", apimgt_url, access_token, verify_ssl)
if not publishSuccessful:
print "[ERROR] API publish failed. [server] %s. Continuing..." % apimgt_url
else:
# Create API
print "\t%s:%s \t :Creating..." % (api_request["name"], api_request["version"])
# Build API Create request
create_req_body = json.loads(open("api_template.json").read())
create_req_body["name"] = api_request["name"]
create_req_body["description"] = api_request["description"]
create_req_body["context"] = api_request["context"]
create_req_body["version"] = api_request["version"]
create_req_body["provider"] = requesting_user
create_req_body["tags"].append(department)
create_req_body["businessInformation"]["businessOwnerEmail"] = requesting_user
create_req_body["businessInformation"]["technicalOwnerEmail"] = requesting_user
create_req_body["businessInformation"]["businessOwner"] = requesting_user
create_req_body["businessInformation"]["technicalOwner"] = requesting_user
create_req_body["status"] = api_status
create_req_body["accessControl"] = "RESTRICTED"
# API creation will fail if role does not exist in userstore
create_req_body["accessControlRoles"].append(department)
create_req_body["visibleRoles"].append(department)
ep_config = json.loads(create_req_body["endpointConfig"])
ep_config["production_endpoints"]["url"] = backend_url_prod
ep_config["sandbox_endpoints"]["url"] = backend_url_sandbox
create_req_body["endpointConfig"] = json.dumps(ep_config)
if "apiDefinition" in api_request:
create_req_body["apiDefinition"] = api_request["apiDefinition"]
else:
swagger_def = json.loads(create_req_body["apiDefinition"])
swagger_def["info"]["title"] = api_request["name"]
swagger_def["info"]["description"] = api_request["description"]
swagger_def["info"]["version"] = api_request["version"]
swagger_def["info"]["contact"]["email"] = requesting_user
swagger_def["info"]["contact"]["name"] = requesting_user
create_req_body["apiDefinition"] = json.dumps(swagger_def)
successful, apiId = api_utils.create_api(create_req_body, apimgt_url, access_token, verify_ssl)
if not successful:
print "[ERROR] API creation failed. [server] %s. Continuing..." % apimgt_url
# Get apiId from create response and then publish
publishSuccessful = api_utils.change_lifecycle(apiId, "Publish", apimgt_url, access_token, verify_ssl)
if not publishSuccessful:
print "[ERROR] API publish failed. [server] %s. Continuing..." % apimgt_url
print
print "DONE!"
exit(0)