class NotificationsUtil(object):
+ """
+ Util for notifications
+ """
+
def __init__(self, notification_type):
self.notification_type = notification_type
self.notifyserializer = None
pass
def send_notification(self):
+ """
+ Send notification
+ :return:
+ """
notification = self.prepare_notification()
subscriptions_filter = {v + "__contains": notification[k] for k, v in self.filter.items()}
self.post_notification(callbackuri, notification)
def post_notification(self, callbackuri, notification, auth_info=None):
+ """
+ Post notification
+ :param callbackuri:
+ :param notification:
+ :param auth_info:
+ :return:
+ """
try:
if auth_info:
if const.BASIC in auth_info.get("authType", ''):
class PkgNotifications(NotificationsUtil):
+ """
+ Notification Utils for VNF pckages
+ """
+
def __init__(self, notification_type, vnf_pkg_id, change_type=None, operational_state=None):
super(PkgNotifications, self).__init__(notification_type)
self.filter = {
self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self):
+ """
+ Prepare notification
+ :return:
+ """
logger.info('Start to prepare Pkgnotification')
vnf_pkg = VnfPackageModel.objects.filter(vnfPackageId=self.vnf_pkg_id)
class NsdNotifications(NotificationsUtil):
+ """
+ Notification Util for NS packages
+ """
+
def __init__(self, notification_type, nsd_info_id, nsd_id, failure_details=None, operational_state=None):
super(NsdNotifications, self).__init__(notification_type)
self.filter = {
# self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self):
+ """
+ Prepare notification
+ :return:
+ """
logger.info('Start to prepare Nsdnotification')
notification_content = {
class PnfNotifications(NotificationsUtil):
+ """
+ Notification util for PNF package
+ """
def __init__(self, notification_type, pnfd_info_id, pnfd_id, failure_details=None):
super(PnfNotifications, self).__init__(notification_type)
self.filter = {
# self.notifyserializer = PkgOnboardingNotificationSerializer
def prepare_notification(self, *args, **kwargs):
+ """
+ Prepare notification
+ :param args:
+ :param kwargs:
+ :return:
+ """
logger.info('Start to prepare Pnfnotification')
notification_content = {
'id': str(uuid.uuid4()), # shall be the same if sent multiple times due to multiple subscriptions.