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