1 # Copyright 2021 Selvi Jayaraman (Tata Elxsi)
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
16 __author__
= "Selvi Jayaraman <selvi.j@tataelxsi.co.in>"
18 from osm_nbi
.subscription_topics
import CommonSubscriptions
19 from osm_nbi
.validation
import vnf_subscription
21 class VnflcmSubscriptionsTopic(CommonSubscriptions
):
22 schema_new
= vnf_subscription
23 def _subscription_mapper(self
, _id
, data
, table
):
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
32 "reference": data
.get("_id"),
33 "CallbackUri": data
.get("CallbackUri")
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"]
58 update_dict
["operationTypes"] = "ANY"
59 if "operationStates" in data
["filter"].keys():
60 update_dict
["operationStates"] = data
["filter"]["operationStates"]
62 update_dict
["operationStates"] = "ANY"
63 formatted_data
.append(update_dict
)
64 self
.db
.create_list(table
, formatted_data
)