HubSpot 用规则引擎重构 JITA 授权：把条件判断变成可观测的决策图

from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass
from typing import Any, Callable


@dataclass
class RuleResult:
    rule: str
    allowed: bool
    reason: str
    duration_ms: float
    dependencies: list[str]


@dataclass
class Rule:
    name: str
    check: Callable[[dict[str, Any]], tuple[bool, str]]
    dependencies: tuple[str, ...] = ()


class RuleEngine:
    def __init__(self, rules: list[Rule]) -> None:
        self.rules = {rule.name: rule for rule in rules}

    def evaluate(self, root: str, request: dict[str, Any]) -> dict[str, Any]:
        results: dict[str, RuleResult] = {}
        visiting: set[str] = set()

        def run(name: str) -> bool:
            if name in results:
                return results[name].allowed
            if name in visiting:
                raise ValueError(f'cycle detected at rule: {name}')
            if name not in self.rules:
                raise KeyError(f'unknown rule: {name}')

            visiting.add(name)
            rule = self.rules[name]
            dependency_states = [run(dep) for dep in rule.dependencies]
            started = time.perf_counter()

            if not all(dependency_states):
                allowed, reason = False, 'DEPENDENCY_DENIED'
            else:
                allowed, reason = rule.check(request)

            duration_ms = (time.perf_counter() - started) * 1000
            results[name] = RuleResult(
                rule=name,
                allowed=allowed,
                reason=reason,
                duration_ms=round(duration_ms, 3),
                dependencies=list(rule.dependencies),
            )
            visiting.remove(name)
            return allowed

        allowed = run(root)
        return {
            'request_id': request['request_id'],
            'decision': 'ALLOW' if allowed else 'DENY',
            'root_rule': root,
            'rule_results': [asdict(result) for result in results.values()],
        }


def pass_rule(_: dict[str, Any]) -> tuple[bool, str]:
    return True, 'ALL_DEPENDENCIES_PASSED'


rules = [
    Rule(
        name='ticket_is_open',
        check=lambda req: (
            req.get('ticket_status') == 'OPEN',
            'TICKET_OPEN' if req.get('ticket_status') == 'OPEN' else 'TICKET_NOT_OPEN',
        ),
    ),
    Rule(
        name='manager_approved',
        check=lambda req: (
            bool(req.get('manager_approved')),
            'APPROVED' if req.get('manager_approved') else 'APPROVAL_MISSING',
        ),
    ),
    Rule(
        name='duration_within_limit',
        check=lambda req: (
            0 < req.get('duration_minutes', 0) <= 60,
            'DURATION_VALID' if 0 < req.get('duration_minutes', 0) <= 60 else 'DURATION_EXCEEDED',
        ),
    ),
    Rule(
        name='production_jita_access',
        check=pass_rule,
        dependencies=('ticket_is_open', 'manager_approved', 'duration_within_limit'),
    ),
]

request = {
    'request_id': 'req-2025-001',
    'ticket_status': 'OPEN',
    'manager_approved': True,
    'duration_minutes': 45,
}

print(json.dumps(RuleEngine(rules).evaluate('production_jita_access', request), indent=2))

python jita_rules.py