hat.event.backends.lmdb.conditions

  1from hat import json
  2
  3from hat.event.backends.lmdb import common
  4
  5
  6class Conditions:
  7
  8    def __init__(self, conf: json.Data):
  9        self._conditions = [
 10            (common.create_subscription(tuple(event_type)
 11                                        for event_type in i['subscriptions']),
 12             _create_condition(i['condition']))
 13            for i in conf]
 14
 15    def matches(self, event: common.Event) -> bool:
 16        for subscription, condition in self._conditions:
 17            if not subscription.matches(event.type):
 18                continue
 19
 20            if not condition.matches(event):
 21                return False
 22
 23        return True
 24
 25
 26def _create_condition(conf):
 27    if conf['type'] == 'all':
 28        return _AllCondition(conf)
 29
 30    if conf['type'] == 'any':
 31        return _AnyCondition(conf)
 32
 33    if conf['type'] == 'json':
 34        return _JsonCondition(conf)
 35
 36    raise ValueError('unsupported condition type')
 37
 38
 39class _AllCondition:
 40
 41    def __init__(self, conf):
 42        self._conditions = [_create_condition(i) for i in conf['conditions']]
 43
 44    def matches(self, event):
 45        return all(condition.matches(event) for condition in self._conditions)
 46
 47
 48class _AnyCondition:
 49
 50    def __init__(self, conf):
 51        self._conditions = [_create_condition(i) for i in conf['conditions']]
 52
 53    def matches(self, event):
 54        return any(condition.matches(event) for condition in self._conditions)
 55
 56
 57class _JsonCondition:
 58
 59    def __init__(self, conf):
 60        self._conf = conf
 61
 62    def matches(self, event):
 63        if not isinstance(event.payload, common.EventPayloadJson):
 64            return False
 65
 66        data_path = self._conf.get('data_path', [])
 67        data = json.get(event.payload.data, data_path)
 68
 69        if 'data_type' in self._conf:
 70            data_type = self._conf['data_type']
 71
 72            if data_type == 'null':
 73                if data is not None:
 74                    return False
 75
 76            elif data_type == 'boolean':
 77                if not isinstance(data, bool):
 78                    return False
 79
 80            elif data_type == 'string':
 81                if not isinstance(data, str):
 82                    return False
 83
 84            elif data_type == 'number':
 85                if not (isinstance(data, float) or
 86                        (isinstance(data, int) and
 87                         not isinstance(data, bool))):
 88                    return False
 89
 90            elif data_type == 'array':
 91                if not isinstance(data, list):
 92                    return False
 93
 94            elif data_type == 'object':
 95                if not isinstance(data, dict):
 96                    return False
 97
 98        if 'data_value' in self._conf:
 99            if self._conf['data_value'] != data:
100                return False
101
102        return True