Loading services/TS29222_CAPIF_Events_API/capif_events/core/notifications.py +48 −7 Original line number Diff line number Diff line Loading @@ -39,17 +39,58 @@ class Notifications(): if EventSubscription.return_supp_feat_dict(sub["supported_features"])["EnhancedEventReport"]: event_detail={} current_app.logger.debug(f"event: {event_detail_redis}") # The nth entry in the "eventFilters" attribute shall correspond to the nth entry in the "events" attribute event_filters = sub.get("event_filters", None) event_filter=None current_app.logger.debug(f"Event filters: {event_filters}") if event_filters: try: event_filter = event_filters[event_filters.index(event)] except IndexError: event_filter=None if event in ["SERVICE_API_AVAILABLE", "SERVICE_API_UNAVAILABLE"]: if event_filter: api_ids_list = event_filter.get("api_ids", None) if api_ids_list and event_detail_redis.get('apiIds', None)[0] in api_ids_list: event_detail["apiIds"]=event_detail_redis.get('apiIds', None) if EventSubscription.return_supp_feat_dict(sub["supported_features"])["ApiStatusMonitoring"]: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) else: event_detail["apiIds"]=event_detail_redis.get('apiIds', None) if EventSubscription.return_supp_feat_dict(sub["supported_features"])["ApiStatusMonitoring"]: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) elif event in ["SERVICE_API_UPDATE"]: if event_filter: api_ids_list = event_filter.get("api_ids", None) if api_ids_list and event_detail_redis.get('apiIds', None)[0] in api_ids_list: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) elif event in ["API_INVOKER_ONBOARDED", "API_INVOKER_OFFBOARDED", "API_INVOKER_UPDATED"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) if invoker_ids_list and event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list: event_detail["apiInvokerIds"]=event_detail_redis.get('apiInvokerIds', None) else: event_detail["apiInvokerIds"]=event_detail_redis.get('apiInvokerIds', None) elif event in ["ACCESS_CONTROL_POLICY_UPDATE"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) api_ids_list = event_filter.get("api_ids", None) if invoker_ids_list and (event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list or event_detail_redis.get('apiIds', None)[0] in api_ids_list): event_detail["accCtrlPolList"]=event_detail_redis.get('accCtrlPolList', None) else: event_detail["accCtrlPolList"]=event_detail_redis.get('accCtrlPolList', None) elif event in ["SERVICE_API_INVOCATION_SUCCESS", "SERVICE_API_INVOCATION_FAILURE"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) api_ids_list = event_filter.get("api_ids", None) aef_ids_list = event_filter.get("aef_ids", None) if invoker_ids_list and (event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list or event_detail_redis.get('apiIds', None)[0] in api_ids_list or event_detail_redis.get('aefIds', None)[0] in aef_ids_list ): event_detail["invocationLogs"]=event_detail_redis.get('invocationLogs', None) else: event_detail["invocationLogs"]=event_detail_redis.get('invocationLogs', None) elif event in ["API_TOPOLOGY_HIDING_CREATED", "API_TOPOLOGY_HIDING_REVOKED"]: event_detail["apiTopoHide"]=event_detail_redis.get('apiTopoHide', None) Loading Loading
services/TS29222_CAPIF_Events_API/capif_events/core/notifications.py +48 −7 Original line number Diff line number Diff line Loading @@ -39,17 +39,58 @@ class Notifications(): if EventSubscription.return_supp_feat_dict(sub["supported_features"])["EnhancedEventReport"]: event_detail={} current_app.logger.debug(f"event: {event_detail_redis}") # The nth entry in the "eventFilters" attribute shall correspond to the nth entry in the "events" attribute event_filters = sub.get("event_filters", None) event_filter=None current_app.logger.debug(f"Event filters: {event_filters}") if event_filters: try: event_filter = event_filters[event_filters.index(event)] except IndexError: event_filter=None if event in ["SERVICE_API_AVAILABLE", "SERVICE_API_UNAVAILABLE"]: if event_filter: api_ids_list = event_filter.get("api_ids", None) if api_ids_list and event_detail_redis.get('apiIds', None)[0] in api_ids_list: event_detail["apiIds"]=event_detail_redis.get('apiIds', None) if EventSubscription.return_supp_feat_dict(sub["supported_features"])["ApiStatusMonitoring"]: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) else: event_detail["apiIds"]=event_detail_redis.get('apiIds', None) if EventSubscription.return_supp_feat_dict(sub["supported_features"])["ApiStatusMonitoring"]: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) elif event in ["SERVICE_API_UPDATE"]: if event_filter: api_ids_list = event_filter.get("api_ids", None) if api_ids_list and event_detail_redis.get('apiIds', None)[0] in api_ids_list: event_detail["serviceAPIDescriptions"]=event_detail_redis.get('serviceAPIDescriptions', None) elif event in ["API_INVOKER_ONBOARDED", "API_INVOKER_OFFBOARDED", "API_INVOKER_UPDATED"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) if invoker_ids_list and event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list: event_detail["apiInvokerIds"]=event_detail_redis.get('apiInvokerIds', None) else: event_detail["apiInvokerIds"]=event_detail_redis.get('apiInvokerIds', None) elif event in ["ACCESS_CONTROL_POLICY_UPDATE"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) api_ids_list = event_filter.get("api_ids", None) if invoker_ids_list and (event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list or event_detail_redis.get('apiIds', None)[0] in api_ids_list): event_detail["accCtrlPolList"]=event_detail_redis.get('accCtrlPolList', None) else: event_detail["accCtrlPolList"]=event_detail_redis.get('accCtrlPolList', None) elif event in ["SERVICE_API_INVOCATION_SUCCESS", "SERVICE_API_INVOCATION_FAILURE"]: if event_filter: invoker_ids_list = event_filter.get("api_invoker_ids", None) api_ids_list = event_filter.get("api_ids", None) aef_ids_list = event_filter.get("aef_ids", None) if invoker_ids_list and (event_detail_redis.get('apiInvokerIds', None)[0] in invoker_ids_list or event_detail_redis.get('apiIds', None)[0] in api_ids_list or event_detail_redis.get('aefIds', None)[0] in aef_ids_list ): event_detail["invocationLogs"]=event_detail_redis.get('invocationLogs', None) else: event_detail["invocationLogs"]=event_detail_redis.get('invocationLogs', None) elif event in ["API_TOPOLOGY_HIDING_CREATED", "API_TOPOLOGY_HIDING_REVOKED"]: event_detail["apiTopoHide"]=event_detail_redis.get('apiTopoHide', None) Loading