-
Notifications
You must be signed in to change notification settings - Fork 7
/
rule_bank_setup.py
99 lines (83 loc) · 3.67 KB
/
rule_bank_setup.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
from datetime import datetime
from sqlalchemy import event, MetaData
from sqlalchemy.engine import Engine
from logic_bank.rule_bank.rule_bank import RuleBank
from logic_bank.rule_bank import rule_bank_withdraw
from logic_bank.exec_trans_logic.listeners import before_flush, before_commit, after_flush
from sqlalchemy.orm import session
import logging
__version__ = "01.20.07" # fix cascade update failure in test_upd_order_shipped
def setup(a_session: session):
"""
Create the RuleBank
Register before_flush listeners
"""
rules_bank = RuleBank()
event.listen(a_session, "before_flush", before_flush)
event.listen(a_session, "before_commit", before_commit)
event.listen(a_session, "after_flush", after_flush)
rules_bank.orm_objects = {}
rules_bank._at = datetime.now()
return rules_bank
def setup_early_row_event_all_classes(early_row_event_all_classes: callable):
ll = RuleBank()
ll._early_row_event_all_classes = early_row_event_all_classes
def set_referring_children(rule, dependency: list):
pass
def compute_formula_execution_order_for_class(class_name: str):
"""
compute formula._exec_order per formula._dependencies
"""
formula_list = rule_bank_withdraw.get_formula_rules(class_name)
formula_list_dict = {}
for each_formula in formula_list:
formula_list_dict[each_formula._column] = each_formula
exec_order = 0
blocked = False
while not blocked and exec_order < len(formula_list):
blocked = True
for each_formula_name in formula_list_dict:
each_formula = formula_list_dict[each_formula_name]
refers_to = ""
if each_formula._exec_order == -1:
for each_referenced_col_name in each_formula._dependencies:
if each_referenced_col_name in formula_list_dict:
referenced_formula = formula_list_dict[each_referenced_col_name]
if referenced_formula._exec_order == -1: # ref'd col done?
if each_referenced_col_name != each_formula_name:
refers_to = referenced_formula._column
break # can't do me yet - ref'd col may also have rules but not yet loaded
if refers_to == "":
exec_order += 1
each_formula._exec_order = exec_order
blocked = False
if blocked:
cycles = ""
cycle_count = 0
for each_formula_name in formula_list_dict:
each_formula = formula_list_dict[each_formula_name]
if each_formula._exec_order == -1:
if cycle_count > 0:
cycles += ", "
cycle_count += 1
cycles += each_formula._column
raise Exception("Mapped Class[" + class_name + "] blocked by circular dependencies:" + cycles)
def compute_formula_execution_order() -> bool:
"""
Determine formula execution order based on "row.xx" references (dependencies),
(or raise exception if cycles detected).
"""
global version
rules_bank = RuleBank()
for each_key in rules_bank.orm_objects:
compute_formula_execution_order_for_class(class_name=each_key)
logic_logger = logging.getLogger("logic_logger")
rule_count = 0
logic_logger.debug(f'\nThe following rules have been activated\n')
list_rules = rules_bank.__str__()
loaded_rules = list(list_rules.split("\n"))
for each_rule in loaded_rules:
logic_logger.debug(each_rule)
rule_count += 1
logic_logger.info(f'Logic Bank {__version__} - {rule_count} rules loaded')
return True