60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Awaitable, Callable, Dict, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class RuleContext:
|
|
data: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
return self.data.get(key, default)
|
|
|
|
def set(self, key: str, value: Any) -> None:
|
|
self.data[key] = value
|
|
|
|
|
|
@dataclass
|
|
class RuleResult:
|
|
matched: bool = False
|
|
stop: bool = False
|
|
action: str = ""
|
|
payload: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
Predicate = Callable[[RuleContext], Awaitable[bool]]
|
|
Action = Callable[[RuleContext], Awaitable[RuleResult]]
|
|
|
|
|
|
@dataclass
|
|
class Rule:
|
|
name: str
|
|
priority: int
|
|
predicate: Predicate
|
|
action: Action
|
|
|
|
|
|
class RuleEngine:
|
|
"""Priority-ordered async rule chain."""
|
|
|
|
def __init__(self, rules: Optional[List[Rule]] = None):
|
|
self._rules: List[Rule] = sorted(rules or [], key=lambda x: x.priority)
|
|
|
|
def add_rule(self, rule: Rule) -> None:
|
|
self._rules.append(rule)
|
|
self._rules.sort(key=lambda x: x.priority)
|
|
|
|
async def run(self, ctx: RuleContext) -> RuleResult:
|
|
for rule in self._rules:
|
|
if not await rule.predicate(ctx):
|
|
continue
|
|
result = await rule.action(ctx)
|
|
if not result.matched:
|
|
result.matched = True
|
|
if not result.action:
|
|
result.action = rule.name
|
|
if result.stop:
|
|
return result
|
|
return RuleResult(matched=False, stop=False, action="no_match")
|