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
22 class VnflcmSubscriptionsTopic(CommonSubscriptions
):
23 schema_new
= vnf_subscription
25 def _subscription_mapper(self
, _id
, data
, table
):
27 Performs data transformation on subscription request
28 :param _id: subscription reference id
29 :param data: data to be transformed
30 :param table: table in which transformed data are inserted
34 "reference": data
.get("_id"),
35 "CallbackUri": data
.get("CallbackUri"),
37 if data
.get("authentication"):
38 formed_data
.update({"authentication": data
.get("authentication")})
39 if data
.get("filter"):
40 if data
["filter"].get("VnfInstanceSubscriptionFilter"):
41 key
= list(data
["filter"]["VnfInstanceSubscriptionFilter"].keys())[0]
42 identifier
= data
["filter"]["VnfInstanceSubscriptionFilter"][key
]
43 formed_data
.update({"identifier": identifier
})
44 if data
["filter"].get("notificationTypes"):
45 for elem
in data
["filter"].get("notificationTypes"):
46 update_dict
= formed_data
.copy()
47 update_dict
["notificationType"] = elem
48 if elem
== "VnfIdentifierCreationNotification":
49 update_dict
["operationTypes"] = "CREATE"
50 update_dict
["operationStates"] = "ANY"
51 formatted_data
.append(update_dict
)
52 elif elem
== "VnfIdentifierDeletionNotification":
53 update_dict
["operationTypes"] = "DELETE"
54 update_dict
["operationStates"] = "ANY"
55 formatted_data
.append(update_dict
)
56 elif elem
== "VnfLcmOperationOccurrenceNotification":
57 if "operationTypes" in data
["filter"].keys():
58 update_dict
["operationTypes"] = data
["filter"][
62 update_dict
["operationTypes"] = "ANY"
63 if "operationStates" in data
["filter"].keys():
64 update_dict
["operationStates"] = data
["filter"][
68 update_dict
["operationStates"] = "ANY"
69 formatted_data
.append(update_dict
)
70 self
.db
.create_list(table
, formatted_data
)