forked from botswana-harvard/edc-metadata
-
Notifications
You must be signed in to change notification settings - Fork 1
/
predicate.py
136 lines (107 loc) · 4.28 KB
/
predicate.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
from __future__ import annotations
from typing import Any
from django.apps import apps as django_apps
from django.core.exceptions import ObjectDoesNotExist
class PredicateError(Exception):
pass
class NoValueError(Exception):
pass
class BasePredicate:
@staticmethod
def get_value(attr: str = None, source_model: str | None = None, **kwargs) -> Any:
"""Returns a value by checking for the attr on each arg.
Each arg in args may be a model instance, queryset, or None.
If not found, does a lookup on the source_model.
"""
found: bool = False
value: Any = None
for v in kwargs.values():
try:
value = getattr(v, attr)
except AttributeError:
continue
else:
found = True
break
if not found:
visit = kwargs.get("visit")
try:
obj = django_apps.get_model(source_model).objects.get(
subject_visit__subject_identifier=visit.subject_identifier,
subject_visit__visit_schedule_name=visit.visit_schedule_name,
subject_visit__schedule_name=visit.schedule_name,
subject_visit__visit_code=visit.visit_code,
subject_visit__visit_code_sequence=visit.visit_code_sequence,
subject_visit__site=visit.site,
)
except ObjectDoesNotExist:
value = None
else:
value = getattr(obj, attr)
return value
class P(BasePredicate):
"""
Simple predicate class.
For example:
predicate = P('gender', 'eq', 'MALE')
predicate = P('referral_datetime', 'is not', None)
predicate = P('age', '<=', 64)
"""
funcs = {
"is": lambda x, y: True if x is y else False,
"is not": lambda x, y: True if x is not y else False,
"gt": lambda x, y: True if x > y else False,
">": lambda x, y: True if x > y else False,
"gte": lambda x, y: True if x >= y else False,
">=": lambda x, y: True if x >= y else False,
"lt": lambda x, y: True if x < y else False,
"<": lambda x, y: True if x < y else False,
"lte": lambda x, y: True if x <= y else False,
"<=": lambda x, y: True if x <= y else False,
"eq": lambda x, y: True if x == y else False,
"equals": lambda x, y: True if x == y else False,
"==": lambda x, y: True if x == y else False,
"neq": lambda x, y: True if x != y else False,
"!=": lambda x, y: True if x != y else False,
"in": lambda x, y: True if x in y else False,
}
def __init__(self, attr: str, operator: str, expected_value: list | str) -> None:
self.attr = attr
self.expected_value = expected_value
self.func = self.funcs.get(operator)
if not self.func:
raise PredicateError(f"Invalid operator. Got {operator}.")
self.operator = operator
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}({self.attr}, {self.operator}, "
f"{self.expected_value})"
)
def __call__(self, **kwargs) -> bool:
value = self.get_value(attr=self.attr, **kwargs)
return self.func(value, self.expected_value)
class PF(BasePredicate):
"""
Predicate with a lambda function.
predicate = PF('age', lambda x: True if x >= 18 and x <= 64 else False)
if lamda is anything more complicated just pass a func directly to the predicate attr:
def my_func(visit, registered_subject, source_obj, source_qs):
if source_obj.married and registered_subject.gender == FEMALE:
return True
return False
class MyRuleGroup(RuleGroup):
my_rule = Rule(
...
predicate = my_func
...
"""
def __init__(self, *attrs, func: callable = None) -> None:
self.attrs = attrs
self.func = func
def __call__(self, **kwargs) -> Any:
values = []
for attr in self.attrs:
values.append(self.get_value(attr=attr, **kwargs))
return self.func(*values)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.attrs}, {self.func})"