near-rt-ric-simulator/ric-plt/a1/main.py

   1 import connexion
   2 import fileinput
   3 import json
   4 import sys
   5
   6 from flask import Flask, escape, request, make_response
   7 from jsonschema import validate
   8 from var_declaration import policy_instances, policy_types, policy_status, policy_type_per_instance
   9
  10 app = connexion.App(__name__, specification_dir='.')
  11
  12 @app.route('/policytypes/<string:policyTypeId>', methods=['PUT','DELETE'])
  13 def policy_type(policyTypeId):
  14   if request.method == 'PUT':
  15     data = request.data.decode("utf-8")
  16     data = data.replace("'", "\"")
  17     data = json.loads(data)
  18     policy_types[policyTypeId] = data
  19     return ('The policy type was either created or updated for policy type id: ' + policyTypeId)
  20   elif request.method == 'DELETE':
  21     if policyTypeId in policy_types.keys():
  22       policy_types.pop(policyTypeId)
  23       return make_response("policy type successfully deleted for policy type id: " + policyTypeId, 200)
  24     else:
  25       return make_response("No policy type defined for the specified id", 404)
  26
  27 @app.route('/', methods=['GET'])
  28 def test():
  29     return("Everything is fine", 200)
  30
  31 @app.route('/deleteinstances', methods=['DELETE'])
  32 def delete_instances():
  33   global policy_instances
  34   global policy_status
  35   global policy_type_per_instance
  36   policy_instances.clear()
  37   policy_status.clear()
  38   policy_type_per_instance.clear()
  39   return("All policy instances deleted", 200)
  40
  41 @app.route('/deletetypes', methods=['DELETE'])
  42 def delete_types():
  43   global policy_types
  44   policy_types.clear()
  45   return("All policy types deleted", 200)
  46
  47 @app.route('/<string:policyId>/<string:enforceStatus>', methods=['PUT'])
  48 def set_status(policyId, enforceStatus):
  49   if policyId in policy_instances.keys():
  50     if policy_type_per_instance[policyId] == "UNDEFINED":
  51       ps = {}
  52       ps["policyId"] = policyId
  53       ps["enforceStatus"] = enforceStatus
  54     else:
  55       policy_type_id = policy_type_per_instance[policyId]
  56       status_schema = policy_types[policy_type_id]["statusSchema"]
  57       ps = {}
  58       ps["policyId"] = policyId
  59       ps["enforceStatus"] = enforceStatus
  60       try:
  61         validate(instance=ps, schema=status_schema)
  62       except:
  63         return(set_error(None, "The json does not validate against the status schema.", 400, None, None, None, None, None))
  64   policy_status.pop(policyId)
  65   policy_status[policyId] = ps
  66   return("Status updated for policy: " + policyId, 200)
  67
  68 @app.route('/<string:policyId>/<string:enforceStatus>/<string:enforceReason>', methods=['PUT'])
  69 def set_status_with_reason(policyId, enforceStatus, enforceReason):
  70   if policyId in policy_instances.keys():
  71     if policy_type_per_instance[policyId] == "UNDEFINED":
  72       ps = {}
  73       ps["policyId"] = policyId
  74       ps["enforceStatus"] = enforceStatus
  75       ps["enforceReason"] = enforceReason
  76     else:
  77       policy_type_id = policy_type_per_instance[policyId]
  78       status_schema = policy_types[policy_type_id]["statusSchema"]
  79       ps = {}
  80       ps["policyId"] = policyId
  81       ps["enforceStatus"] = enforceStatus
  82       ps["enforceReason"] = enforceReason
  83       try:
  84         validate(instance=ps, schema=status_schema)
  85       except:
  86         return(set_error(None, "The json does not validate against the status schema.", 400, None, None, None, None, None))
  87   policy_status.pop(policyId)
  88   policy_status[policyId] = ps
  89   return("Status updated for policy: " + policyId, 200)
  90
  91 port_number = 8085
  92 if len(sys.argv) >= 2:
  93   if isinstance(sys.argv[1], int):
  94     port_number = sys.argv[1]
  95
  96 app.add_api('a1-openapi.yaml')
  97 app.run(port=port_number)
  98