1
|
|
|
from abc import ABC, abstractmethod |
2
|
|
|
from typing import List |
3
|
|
|
|
4
|
|
|
from decision_engine.comparisons import Comparison |
5
|
|
|
from decision_engine.sources import Source |
6
|
|
|
|
7
|
|
|
|
8
|
|
|
class Rule(ABC): |
9
|
|
|
def __init__(self, name: str = None): |
10
|
|
|
self.name = name |
11
|
|
|
|
12
|
|
|
@abstractmethod |
13
|
|
|
def __repr__(self): |
14
|
|
|
raise NotImplementedError # pragma: no cover |
15
|
|
|
|
16
|
|
|
@abstractmethod |
17
|
|
|
def check(self, data: dict) -> bool: |
18
|
|
|
raise NotImplementedError # pragma: no cover |
19
|
|
|
|
20
|
|
|
|
21
|
|
|
class SimpleComparisonRule(Rule): |
22
|
|
|
def __init__(self, source1: Source, source2: Source, |
23
|
|
|
comparison: Comparison, name: str = None) -> None: |
24
|
|
|
self.source1 = source1 |
25
|
|
|
self.source2 = source2 |
26
|
|
|
self.comparison = comparison |
27
|
|
|
super().__init__(name) |
28
|
|
|
|
29
|
|
|
def __repr__(self): |
30
|
|
|
return f"Name: '{self.name}' | source1: '{self.source1.name}' | " \ |
31
|
|
|
f"source2: '{self.source2.name}' | " \ |
32
|
|
|
f"comparison: '{self.comparison.__class__.__name__}'" |
33
|
|
|
|
34
|
|
|
def check(self, data: dict) -> bool: |
35
|
|
|
val1 = self.source1.get_value(data) |
36
|
|
|
val2 = self.source2.get_value(data) |
37
|
|
|
return self.comparison.check(val1, val2) |
38
|
|
|
|
39
|
|
|
|
40
|
|
|
class BooleanOrRule(Rule): |
41
|
|
|
def __init__(self, rules: List[Rule], name: str = None) -> None: |
42
|
|
|
self.rules = rules |
43
|
|
|
super().__init__(name) |
44
|
|
|
|
45
|
|
|
def __repr__(self): |
46
|
|
|
return f"Name: '{self.name}' | rules: {self.rules}" |
47
|
|
|
|
48
|
|
|
def check(self, data: dict) -> bool: |
49
|
|
|
return any([rule.check(data) for rule in self.rules]) |
50
|
|
|
|
51
|
|
|
|
52
|
|
|
class BooleanAndRule(Rule): |
53
|
|
|
def __init__(self, rules: List[Rule], name: str = None) -> None: |
54
|
|
|
self.rules = rules |
55
|
|
|
super().__init__(name) |
56
|
|
|
|
57
|
|
|
def __repr__(self): |
58
|
|
|
return f"Name: '{self.name}' | rules: {self.rules}" |
59
|
|
|
|
60
|
|
|
def check(self, data: dict) -> bool: |
61
|
|
|
return all([rule.check(data) for rule in self.rules]) |
62
|
|
|
|