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 var_declaration import policy_instances, policy_types, policy_status
   8
   9 app = connexion.App(__name__, specification_dir='.')
  10
  11 @app.route('/policytypes/<string:policyTypeId>', methods=['PUT','DELETE'])
  12 def policy_type(policyTypeId):
  13   if request.method == 'PUT':
  14     data = request.data.decode("utf-8")
  15     data = data.replace("'", "\"")
  16     data = json.loads(data)
  17     policy_types[policyTypeId] = data
  18     return ('The policy type was either created or updated for policy type id: ' + policyTypeId)
  19   elif request.method == 'DELETE':
  20     if policyTypeId in policy_types.keys():
  21       policy_types.pop(policyTypeId)
  22       return make_response("policy type successfully deleted for policy type id: " + policyTypeId, 200)
  23     else:
  24       return make_response("No policy type defined for the specified id", 404)
  25
  26 @app.route('/', methods=['GET'])
  27 def test():
  28   if "STD_QoSNud_0.1.0" in policy_types.keys():
  29     return("Something fishy!", 200)
  30   else:
  31     return(str(list(policy_types.keys())), 200)
  32
  33 @app.route('/deleteinstances', methods=['DELETE'])
  34 def delete_instances():
  35   global policy_instances
  36   global policy_status
  37   policy_instances.clear()
  38   policy_status.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 enforceStatus in ["UNDEFINED", "ENFORCED", "NOT_ENFORCED"]:
  51       policy_status.pop(policyId)
  52       ps = {}
  53       ps["policyId"] = policyId
  54       ps["enforceStatus"] = enforceStatus
  55       policy_status[policyId] = ps
  56       return("Status updated for policy: " + policyId, 200)
  57     else:
  58       return("enforceStatus should be one of \"UNDEFINED\", \"ENFORCED\" or \"NOT_ENFORCED\"", 400)
  59   else:
  60     return("The policy id does not correspond to any existing policy instance", 400)
  61
  62 @app.route('/<string:policyId>/<string:enforceStatus>/<string:enforceReason>', methods=['PUT'])
  63 def set_status_with_reason(policyId, enforceStatus, enforceReason):
  64   if policyId in policy_instances.keys():
  65     if enforceStatus == "NOT_ENFORCED":
  66       if enforceReason in ["100", "200", "300", "800"]:
  67         policy_status.pop(policyId)
  68         ps = {}
  69         ps["policyId"] = policyId
  70         ps["enforceStatus"] = enforceStatus
  71         ps["enforceReason"] = enforceReason
  72         policy_status[policyId] = ps
  73         return("Status updated for policy: " + policyId, 200)
  74       else:
  75         return("enforceReason should be one of \"100\", \"200\", \"300\" or \"800\"", 400)
  76     else:
  77       return("A status provided together with an enforcement reason should be \"NOT_ENFORCED\"", 400)
  78   else:
  79     return("The policy id does not correspond to any existing policy instance", 404)
  80
  81
  82 port_number = 8085
  83 if len(sys.argv) >= 2:
  84   if isinstance(sys.argv[1], int):
  85     port_number = sys.argv[1]
  86
  87 app.add_api('a1-openapi.yaml')
  88 app.run(port=port_number)
  89