osm_nbi/osm_vnfm/vnf_subscription.py

   1 # Copyright 2021 Selvi Jayaraman (Tata Elxsi)
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 __author__ = "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
  17
  18 from osm_nbi.subscription_topics import CommonSubscriptions
  19 from osm_nbi.validation import vnf_subscription
  20
  21 class VnflcmSubscriptionsTopic(CommonSubscriptions):
  22     schema_new = vnf_subscription
  23     def _subscription_mapper(self, _id, data, table):
  24         """
  25         Performs data transformation on subscription request
  26         :param _id: subscription reference id
  27         :param data: data to be transformed
  28         :param table: table in which transformed data are inserted
  29         """
  30         formatted_data = []
  31         formed_data = {
  32             "reference": data.get("_id"),
  33             "CallbackUri": data.get("CallbackUri")
  34         }
  35         if data.get("authentication"):
  36             formed_data.update({"authentication": data.get("authentication")})
  37         if data.get("filter"):
  38             if data["filter"].get("VnfInstanceSubscriptionFilter"):
  39                 key = list(data["filter"]["VnfInstanceSubscriptionFilter"].keys())[0]
  40                 identifier = data["filter"]["VnfInstanceSubscriptionFilter"][key]
  41                 formed_data.update({"identifier": identifier})
  42             if data["filter"].get("notificationTypes"):
  43                 for elem in data["filter"].get("notificationTypes"):
  44                     update_dict = formed_data.copy()
  45                     update_dict["notificationType"] = elem
  46                     if elem == "VnfIdentifierCreationNotification":
  47                         update_dict["operationTypes"] = "CREATE"
  48                         update_dict["operationStates"] = "ANY"
  49                         formatted_data.append(update_dict)
  50                     elif elem == "VnfIdentifierDeletionNotification":
  51                         update_dict["operationTypes"] = "DELETE"
  52                         update_dict["operationStates"] = "ANY"
  53                         formatted_data.append(update_dict)
  54                     elif elem == "VnfLcmOperationOccurrenceNotification":
  55                         if "operationTypes" in data["filter"].keys():
  56                             update_dict["operationTypes"] = data["filter"]["operationTypes"]
  57                         else:
  58                             update_dict["operationTypes"] = "ANY"
  59                         if "operationStates" in data["filter"].keys():
  60                             update_dict["operationStates"] = data["filter"]["operationStates"]
  61                         else:
  62                             update_dict["operationStates"] = "ANY"
  63                         formatted_data.append(update_dict)
  64         self.db.create_list(table, formatted_data)
  65         return None