Class SpendControl

uncommon_route/spend_control.py:125–289

Time-windowed spending limiter with persistent storage.

class SpendControl:
    """Time-windowed spending limiter with persistent storage."""

    def __init__(
        self,
        storage: SpendControlStorage | None = None,
        now_fn: Any = None,
    ) -> None:
        self._storage = storage or FileSpendControlStorage()
        self._now = now_fn or time.time
        self._limits = SpendLimits()
        self._history: list[SpendRecord] = []
        self._session_spent: float = 0.0
        self._session_calls: int = 0
        self._load()

    def set_limit(self, window: SpendWindow, amount: float) -> None:
        if amount <= 0 or not isinstance(amount, (int, float)):
            raise ValueError("Limit must be a positive number")
        setattr(self._limits, window, float(amount))
        self._save()

    def clear_limit(self, window: SpendWindow) -> None:
        setattr(self._limits, window, None)
        self._save()

    @property
    def limits(self) -> SpendLimits:
        return self._limits

    def check(self, estimated_cost: float, additional_committed: float = 0.0) -> CheckResult:
        """Check whether ``estimated_cost`` fits under every configured window.

        ``additional_committed`` covers in-flight reservations (estimated costs of
        concurrent requests that passed ``check`` but have not yet been settled).
        It is added to every windowed denominator but NOT to the per-request
        comparison, which stays strictly about a single call.
        """
        now = self._now()

        if self._limits.per_request is not None and estimated_cost > self._limits.per_request:
            return CheckResult(
                allowed=False,
                blocked_by="per_request",
                remaining=self._limits.per_request,
                reason=f"Per-request limit: ${estimated_cost:.4f} > ${self._limits.per_request:.2f} max",
            )

        if self._limits.hourly is not None:
            hourly_spent = self._window_total(now - HOUR_S, now) + additional_committed
            remaining = self._limits.hourly - hourly_spent
            if estimated_cost > remaining:
                oldest = next((r for r in self._history if r.timestamp >= now - HOUR_S), None)
                reset_in = int(oldest.timestamp + HOUR_S - now) if oldest else 0
                return CheckResult(
                    allowed=False, blocked_by="hourly", remaining=remaining,
                    reason=f"Hourly limit: ${hourly_spent + estimated_cost:.2f} > ${self._limits.hourly:.2f}",
                    reset_in_s=max(0, reset_in),