Skip to content

PolicyEngine

Stateful engine that evaluates spend requests against policy rules.

paygraph.policy.PolicyEngine

Stateful engine that evaluates spend requests against policy rules.

Tracks cumulative daily spend in memory. The daily counter resets automatically at the start of each new calendar day.

Source code in src/paygraph/policy.py 
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class PolicyEngine:
    """Stateful engine that evaluates spend requests against policy rules.

    Tracks cumulative daily spend in memory. The daily counter resets
    automatically at the start of each new calendar day.
    """

    def __init__(self, policy: SpendPolicy) -> None:
        """Initialize the engine with a spend policy.

        Args:
            policy: The ``SpendPolicy`` defining governance rules.
        """
        self.policy = policy
        self._daily_spend: float = 0.0
        self._current_date: date = date.today()

    def _reset_daily_if_needed(self) -> None:
        today = date.today()
        if today != self._current_date:
            self._daily_spend = 0.0
            self._current_date = today

    def evaluate(
        self,
        amount: float,
        vendor: str,
        justification: str | None = None,
        on_check: Callable[[str, bool], None] | None = None,
    ) -> PolicyResult:
        """Evaluate a spend request against all policy rules.

        Checks are run in order: amount_cap, vendor_allowlist,
        vendor_blocklist, mcc_filter, daily_budget, justification.
        Evaluation stops at the first failure.

        Args:
            amount: Dollar amount of the spend request.
            vendor: Name of the vendor or service.
            justification: Reason for the spend (required if
                ``policy.require_justification`` is True).
            on_check: Optional callback invoked after each check with
                ``(check_name, passed)``.

        Returns:
            A ``PolicyResult`` indicating approval or denial.
        """
        self._reset_daily_if_needed()
        checks_passed: list[str] = []

        def _pass(name: str) -> None:
            checks_passed.append(name)
            if on_check:
                on_check(name, True)

        def _fail(name: str, reason: str) -> PolicyResult:
            if on_check:
                on_check(name, False)
            return PolicyResult(
                approved=False,
                denial_reason=reason,
                checks_passed=checks_passed,
            )

        # 1. Amount cap
        if amount > self.policy.max_transaction:
            return _fail(
                "amount_cap",
                f"Amount ${amount:.2f} exceeds limit of ${self.policy.max_transaction:.2f}",
            )
        _pass("amount_cap")

        # 2. Vendor allowlist / blocklist
        vendor_lower = vendor.lower()
        if self.policy.allowed_vendors is not None:
            if not any(v.lower() in vendor_lower for v in self.policy.allowed_vendors):
                return _fail("vendor_allowlist", f"Vendor '{vendor}' is not in the allowed list")
        _pass("vendor_allowlist")

        if self.policy.blocked_vendors is not None:
            if any(v.lower() in vendor_lower for v in self.policy.blocked_vendors):
                return _fail("vendor_blocklist", f"Vendor '{vendor}' is blocked")
        _pass("vendor_blocklist")

        # 3. MCC filter (stubbed — no MCC in spend request yet)
        _pass("mcc_filter")

        # 4. Daily budget
        if self._daily_spend + amount > self.policy.daily_budget:
            return _fail(
                "daily_budget",
                f"Daily budget exhausted (${self._daily_spend:.2f} / ${self.policy.daily_budget:.2f})",
            )
        _pass("daily_budget")

        # 5. Justification present
        if self.policy.require_justification and not justification:
            return _fail("justification", "Justification is required but was not provided")
        _pass("justification")

        # Approved — increment daily spend
        self._daily_spend += amount

        return PolicyResult(approved=True, checks_passed=checks_passed)

__init__(policy)

Initialize the engine with a spend policy.

Parameters:

Name Type Description Default
policy SpendPolicy

The SpendPolicy defining governance rules.

 required
Source code in src/paygraph/policy.py 
53
54
55
56
57
58
59
60
61
def __init__(self, policy: SpendPolicy) -> None:
    """Initialize the engine with a spend policy.

    Args:
        policy: The ``SpendPolicy`` defining governance rules.
    """
    self.policy = policy
    self._daily_spend: float = 0.0
    self._current_date: date = date.today()

evaluate(amount, vendor, justification=None, on_check=None)

Evaluate a spend request against all policy rules.

Checks are run in order: amount_cap, vendor_allowlist, vendor_blocklist, mcc_filter, daily_budget, justification. Evaluation stops at the first failure.

Parameters:

Name Type Description Default
amount float

Dollar amount of the spend request.

 required
vendor str

Name of the vendor or service.

 required
justification str | None

Reason for the spend (required if policy.require_justification is True).

 None
on_check Callable[[str, bool], None] | None

Optional callback invoked after each check with (check_name, passed).

 None

Returns:

Type Description
PolicyResult

A PolicyResult indicating approval or denial.
Source code in src/paygraph/policy.py 
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def evaluate(
    self,
    amount: float,
    vendor: str,
    justification: str | None = None,
    on_check: Callable[[str, bool], None] | None = None,
) -> PolicyResult:
    """Evaluate a spend request against all policy rules.

    Checks are run in order: amount_cap, vendor_allowlist,
    vendor_blocklist, mcc_filter, daily_budget, justification.
    Evaluation stops at the first failure.

    Args:
        amount: Dollar amount of the spend request.
        vendor: Name of the vendor or service.
        justification: Reason for the spend (required if
            ``policy.require_justification`` is True).
        on_check: Optional callback invoked after each check with
            ``(check_name, passed)``.

    Returns:
        A ``PolicyResult`` indicating approval or denial.
    """
    self._reset_daily_if_needed()
    checks_passed: list[str] = []

    def _pass(name: str) -> None:
        checks_passed.append(name)
        if on_check:
            on_check(name, True)

    def _fail(name: str, reason: str) -> PolicyResult:
        if on_check:
            on_check(name, False)
        return PolicyResult(
            approved=False,
            denial_reason=reason,
            checks_passed=checks_passed,
        )

    # 1. Amount cap
    if amount > self.policy.max_transaction:
        return _fail(
            "amount_cap",
            f"Amount ${amount:.2f} exceeds limit of ${self.policy.max_transaction:.2f}",
        )
    _pass("amount_cap")

    # 2. Vendor allowlist / blocklist
    vendor_lower = vendor.lower()
    if self.policy.allowed_vendors is not None:
        if not any(v.lower() in vendor_lower for v in self.policy.allowed_vendors):
            return _fail("vendor_allowlist", f"Vendor '{vendor}' is not in the allowed list")
    _pass("vendor_allowlist")

    if self.policy.blocked_vendors is not None:
        if any(v.lower() in vendor_lower for v in self.policy.blocked_vendors):
            return _fail("vendor_blocklist", f"Vendor '{vendor}' is blocked")
    _pass("vendor_blocklist")

    # 3. MCC filter (stubbed — no MCC in spend request yet)
    _pass("mcc_filter")

    # 4. Daily budget
    if self._daily_spend + amount > self.policy.daily_budget:
        return _fail(
            "daily_budget",
            f"Daily budget exhausted (${self._daily_spend:.2f} / ${self.policy.daily_budget:.2f})",
        )
    _pass("daily_budget")

    # 5. Justification present
    if self.policy.require_justification and not justification:
        return _fail("justification", "Justification is required but was not provided")
    _pass("justification")

    # Approved — increment daily spend
    self._daily_spend += amount

    return PolicyResult(approved=True, checks_passed=checks_passed)