forked from botswana-harvard/edc-metadata
-
Notifications
You must be signed in to change notification settings - Fork 1
/
rule_evaluator.py
79 lines (59 loc) · 2.52 KB
/
rule_evaluator.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
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Type
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from edc_registration import get_registered_subject_model_cls
from ..constants import DO_NOTHING
if TYPE_CHECKING:
from edc_registration.models import RegisteredSubject
from edc_visit_tracking.model_mixins import VisitModelMixin as Base
from ..model_mixins.creates import CreatesMetadataModelMixin
from .logic import Logic
class RelatedVisitModel(CreatesMetadataModelMixin, Base):
pass
class RuleEvaluatorError(Exception):
pass
class RuleEvaluatorRegisterSubjectError(Exception):
pass
show_edc_metadata_warnings = getattr(settings, "EDC_METADATA_SHOW_NOVALUEERROR_WARNING", False)
class RuleEvaluator:
"""A class to evaluate a rule.
Sets `self.result` to REQUIRED, NOT_REQUIRED or None.
Set as a class attribute on Rule.
See also RuleGroup and its metaclass.
"""
def __init__(
self, logic: Logic = None, related_visit: RelatedVisitModel = None, **kwargs
) -> None:
self._registered_subject: RegisteredSubject | None = None
self.logic: Logic = logic
self.result: str | None = None
self.related_visit = related_visit
options = dict(
visit=self.related_visit, registered_subject=self.registered_subject, **kwargs
)
predicate = self.logic.predicate(**options)
if predicate:
if self.logic.consequence != DO_NOTHING:
self.result = self.logic.consequence
else:
if self.logic.alternative != DO_NOTHING:
self.result = self.logic.alternative
@property
def registered_subject_model(self) -> Type[RegisteredSubject]:
return get_registered_subject_model_cls()
@property
def registered_subject(self) -> Any:
"""Returns a registered subject model instance or raises."""
if not self._registered_subject:
try:
self._registered_subject = self.registered_subject_model.objects.get(
subject_identifier=self.related_visit.subject_identifier
)
except ObjectDoesNotExist as e:
raise RuleEvaluatorRegisterSubjectError(
f"Registered subject required for rule {repr(self)}. "
f"subject_identifier='{self.related_visit.subject_identifier}'. "
f"Got {e}."
)
return self._registered_subject