1 # ============LICENSE_START===============================================
2 # Copyright (C) 2022 Nordix Foundation. All rights reserved.
3 # ========================================================================
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 # ============LICENSE_END=================================================
25 from flask import Flask, escape, request, Response, make_response
26 from jsonschema import validate
27 from var_declaration import a1_policy_instances, forced_settings
31 APPL_JSON='application/json'
32 APPL_PROB_JSON='application/problem+json'
34 #Python implementation of EXT_SRV_api.yaml - Open API -
36 # API Function: Get all a1 policy ids
37 def get_all_a1_policies():
39 if ((r := check_modified_response()) is not None):
42 res = list(a1_policy_instances)
45 # API Function: Get A1 policy
46 def get_a1_policy(a1policyId):
48 if ((r := check_modified_response()) is not None):
51 a1_policy_id=str(a1policyId)
53 policykeys=a1_policy_instances.keys()
54 if (a1_policy_id not in policykeys):
55 pjson=create_problem_json(None, "The A1 policy requested does not exist.", 404, None, a1_policy_id)
56 return Response(json.dumps(pjson), 404, mimetype=APPL_PROB_JSON)
58 return Response(json.dumps(a1_policy_instances[a1_policy_id]), 200, mimetype=APPL_JSON)
60 # API Function: Create or update a a1policy
61 def put_a1_policy(a1policyId):
63 if ((r := check_modified_response()) is not None):
66 a1_policy_id=str(a1policyId)
70 data = json.loads(data)
72 pjson=create_problem_json(None, "The a1policy is corrupt or missing.", 400, None, a1_policy_id)
73 return Response(json.dumps(pjson), 400, mimetype=APPL_PROB_JSON)
75 policykeys=a1_policy_instances.keys()
78 if a1_policy_id in policykeys:
81 a1_policy_instances[a1_policy_id]=data
84 return Response(json.dumps(data), 200, mimetype=APPL_JSON)
87 headers['Location']='/a1policy/' + a1_policy_id
88 return Response(json.dumps(data), 201, headers=headers, mimetype=APPL_JSON)
90 # API Function: Delete a a1policy
91 def delete_a1_policy(a1policyId):
93 if ((r := check_modified_response()) is not None):
96 a1_policy_id=str(a1policyId)
98 policykeys=a1_policy_instances.keys()
99 if (a1_policy_id not in policykeys):
100 pjson=create_problem_json(None, "The a1policy does not exist.", 404, None, a1_policy_id)
101 return Response(json.dumps(pjson), 404, mimetype=APPL_PROB_JSON)
103 del a1_policy_instances[a1_policy_id]
104 return Response('', 204, mimetype=APPL_JSON)
107 # Helper: Create a response object if forced http response code is set
108 def get_forced_response():
109 if (forced_settings['code'] is not None):
110 pjson=create_error_response(forced_settings['code'])
111 return Response(json.dumps(pjson), pjson['status'], mimetype=APPL_PROB_JSON)
115 # Helper: Delay if delayed response code is set
117 if (forced_settings['delay'] is not None):
119 val=int(forced_settings['delay'])
125 # Helper: Check if response shall be delayed or a forced response shall be sent
126 def check_modified_response():
128 return get_forced_response()
131 # Helper: Create a problem json object
132 def create_problem_json(type_of, title, status, detail, instance):
134 if type_of is not None:
135 error["type"] = type_of
136 if title is not None:
137 error["title"] = title
138 if status is not None:
139 error["status"] = status
140 if detail is not None:
141 error["detail"] = detail
142 if instance is not None:
143 error["instance"] = instance
147 # Helper: Create a problem json based on a generic http response code
148 def create_error_response(code):
150 return(create_problem_json(None, "Bad request", 400, "Object in payload not properly formulated or not related to the method", None))
152 return(create_problem_json(None, "Not found", 404, "No resource found at the URI", None))
154 return(create_problem_json(None, "Method not allowed", 405, "Method not allowed for the URI", None))
156 return(create_problem_json(None, "Conflict", 409, "Request could not be processed in the current state of the resource", None))
158 return(create_problem_json(None, "Too many requests", 429, "Too many requests have been sent in a given amount of time", None))
160 return(create_problem_json(None, "Insufficient storage", 507, "The method could not be performed on the resource because the provider is unable to store the representation needed to successfully complete the request", None))
162 return(create_problem_json(None, "Service unavailable", 503, "The provider is currently unable to handle the request due to a temporary overload", None))
164 return(create_problem_json(None, "Unknown", code, "Not implemented response code", None))