Collecting context

We have a solid base for a rule engine, let's add some structure to it. Instead of returning a boolean, let's now make sure every rule returns a RuleEngineResult. This will hold not only the result, but also the reason. Which will be a string or empty.

@dataclass
class RuleEngineResult:
    result: bool
    reason: Optional[str]

Let's also create a abstract rule which all Rules inherit and a RuleOp enum

facts_type = dict[str, Union[str, int, float]]
 
class AbstractRule(ABC):
    @abstractmethod
    def run(self, facts: facts_type) -> RuleEngineResult:
        pass
 
class RuleOp(Enum):
    EQ = "EQ"
    GTE = "GTE"

Let's flush out the Rule. Make it inherit the AbstractRule created and re-write it to return a RuleEngineResult

@dataclass
class Rule(AbstractRule):
    fact_key: str | int
    operator: RuleOp
    value: str | int
    reason: Optional[str]
 
    def run(self, facts) -> RuleEngineResult:
        if self.operator == RuleOp.EQ:
            rule_res = facts.get(self.fact_key) == self.value
            return RuleEngineResult(
                result=rule_res,
                reason=self.reason if not rule_res else None,
            )
        elif self.operator == RuleOp.GTE:
            rule_res = facts.get(self.fact_key) >= self.value
            return RuleEngineResult(
                result=rule_res,
                reason=self.reason if not rule_res else None,
            )
        else:
            raise Exception(f"unknown operator: {self.operator}")

Now, the AndRule can be re-written in such a way that we return the result along with the reason if it exists.

@dataclass
class AndRule(AbstractRule):
    rules: list[AbstractRule]
    reason: Optional[str]
 
    def run(self, facts) -> RuleEngineResult:
        for r in self.rules:
            r_res: RuleEngineResult = r.run(facts)
            if not r_res.result:
                # one condition failed, short circuit
                return RuleEngineResult(
                    result=r_res.result,
                    reason=r_res.reason if r_res.reason else self.reason,
                )
        return RuleEngineResult(result=True, reason=None)

With the OrRule, its more tricky. An or rule is a success if one of the condition succeeds. But is a failure if both fail. So technically, the only way to pass context is to pass all the reasons. We'll take a design decision and bubble up the reason of the OrRule, instead of the individual reasons.

@dataclass
class OrRule(AbstractRule):
    rules: list[AbstractRule]
    reason: Optional[str]
 
    def run(self, facts) -> RuleEngineResult:
        for r in self.rules:
            r_res = r.run(facts)
            if r_res.result:
                # one condition satisfied, short circuit
                return RuleEngineResult(
                    result=True, reason=r_res.reason if r_res.reason else self.reason
                )
        # nothing was true, or rule failed
        return RuleEngineResult(result=False, reason=self.reason)

Let's wrap up all of this with a run_engine function.

def run_engine(rule: AbstractRule, facts: facts_type) -> RuleEngineResult:
    return rule.run(facts=facts)

Let's run a quick test

facts_dict: facts_type = {"age": 18, "country": "IN"}
 
USER_UNDER_AGE_RULE = Rule(
    fact_key="age", operator=RuleOp.GTE, value=18, reason="user under min age"
)
USER_IN_US = Rule(fact_key="country", operator=RuleOp.EQ, value="US", reason=None)
USER_IN_IN = OrRule(
    [
        Rule(fact_key="country", operator=RuleOp.EQ, value="IN", reason=None),
        Rule(fact_key="country", operator=RuleOp.EQ, value="india", reason=None),
    ],
    reason=None,
)
USER_IN_ALLOWED_COUNTRIES = OrRule(
    [USER_IN_IN, USER_IN_US], reason="user not in allowed countries"
)
 
PRODUCT_ELIGIBILITY = AndRule(
    [
        USER_UNDER_AGE_RULE,
        USER_IN_ALLOWED_COUNTRIES,
    ],
    reason="user ineligible",
)
 
print(run_engine(rule=PRODUCT_ELIGIBILITY, facts={"age": 18, "country": "RU"})) # RuleEngineResult(result=False, reason='user not in allowed countries')
print(run_engine(rule=PRODUCT_ELIGIBILITY, facts={"age": 18, "country": "IN"})) # RuleEngineResult(result=True, reason=None)

The rule we set out to implement is now complete!

A Simple Rule Wrapping up the rule engine