Source code for watcher.decision_engine.model.notification.filtering
# -*- encoding: utf-8 -*-
# Copyright (c) 2016 b<>com
#
# Authors: Vincent FRANCOISE <Vincent.FRANCOISE@b-com.com>
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re
import oslo_messaging as om
import six
[docs]class NotificationFilter(om.NotificationFilter):
"""Notification Endpoint base class
This class is responsible for handling incoming notifications. Depending
on the priority level of the incoming, you may need to implement one or
more of the following methods:
.. code: py
def audit(self, ctxt, publisher_id, event_type, payload, metadata):
do_something(payload)
def info(self, ctxt, publisher_id, event_type, payload, metadata):
do_something(payload)
def warn(self, ctxt, publisher_id, event_type, payload, metadata):
do_something(payload)
def error(self, ctxt, publisher_id, event_type, payload, metadata):
do_something(payload)
def critical(self, ctxt, publisher_id, event_type, payload, metadata):
do_something(payload)
"""
def _build_regex_dict(self, regex_list):
if regex_list is None:
return {}
regex_mapping = {}
for key, value in regex_list.items():
if isinstance(value, dict):
regex_mapping[key] = self._build_regex_dict(value)
else:
if callable(value):
regex_mapping[key] = value
elif value is not None:
regex_mapping[key] = re.compile(value)
else:
regex_mapping[key] = None
return regex_mapping
def _check_for_mismatch(self, data, regex):
if isinstance(regex, dict):
mismatch_results = [
k not in data or not self._check_for_mismatch(data[k], v)
for k, v in regex.items()
]
if not mismatch_results:
return False
return all(mismatch_results)
elif callable(regex):
# The filter is a callable that should return True
# if there is a mismatch
return regex(data)
elif regex is not None and data is None:
return True
elif (regex is not None and
isinstance(data, six.string_types) and
not regex.match(data)):
return True
return False