class NotificationBase:
-
response_models = None
# Common HTTP payload header for all notifications.
payload_header = {"Content-Type": "application/json", "Accept": "application/json"}
class NsLcmNotification(NotificationBase):
-
# SOL005 response model for nslcm notifications
response_models = {
"NsLcmOperationOccurrenceNotification": {
"NsLcmOperationOccurrenceNotification",
"NsChangeNotification",
"NsIdentifierCreationNotification",
- "NsIdentifierDeletionNotification"
+ "NsIdentifierDeletionNotification",
]
filter_q = {
"identifier": [nsd_id, ns_instance_id],
"operationStates": ["ANY"],
"operationTypes": ["ANY"],
- "notificationType": notification_type
- }
+ "notificationType": notification_type,
+ }
if op_state:
filter_q["operationStates"].append(op_state)
if command:
"changedExtConnectivity",
"modificationsTriggeredByVnfPkgChange",
"error",
- "_links"
+ "_links",
},
"VnfIdentifierCreationNotification": {
"id",
"subscriptionId",
"timeStamp",
"vnfInstanceId",
- "_links"
+ "_links",
},
"VnfIdentifierDeletionNotification": {
"id",
"subscriptionId",
"timeStamp",
"vnfInstanceId",
- "_links"
+ "_links",
},
}
"""
return self.response_models
- def _format_vnflcm_subscribers(self, subscribers: list, event_details: dict) -> list:
+ def _format_vnflcm_subscribers(
+ self, subscribers: list, event_details: dict
+ ) -> list:
"""
Formats the raw event details from kafka message and subscriber details.
:param subscribers: A list of subscribers whom the event needs to be notified.
subscriber.update(event_details["params"])
return subscribers
- def get_subscribers(self, vnfd_id: str, vnf_instance_id: str, command: str, op_state: str,
- event_details: dict) -> list:
+ def get_subscribers(
+ self,
+ vnfd_id: str,
+ vnf_instance_id: str,
+ command: str,
+ op_state: str,
+ event_details: dict,
+ ) -> list:
"""
Queries database and returns list of subscribers.
:param vnfd_id: Vnfd id of a VNF whose lifecycle has changed. (instantiated, scaled, terminated. etc)
notification_type = [
"VnfIdentifierCreationNotification",
"VnfLcmOperationOccurrenceNotification",
- "VnfIdentifierDeletionNotification"
+ "VnfIdentifierDeletionNotification",
]
filter_q = {
"identifier": [vnfd_id, vnf_instance_id],
"operationStates": ["ANY"],
"operationTypes": ["ANY"],
- "notificationType": notification_type
+ "notificationType": notification_type,
}
if op_state:
filter_q["operationStates"].append(op_state)