11 - Custom Store
bursar ships with two store implementations: PostgresStore (production-ready, persistent) and MemoryStore (development, ephemeral). But your application might use a different backend -- Redis for speed, DynamoDB for scalability, SQLite for embedded deployments. The CreditStore abstract base class (ABC) defines the contract that every store must fulfill.
Think of the CreditStore ABC as a standardized electrical outlet. The shape of the outlet (the abstract methods) is the same everywhere, but what happens behind the wall (the implementation) can be anything -- Postgres, Redis, DynamoDB, or a plain Python dictionary. As long as the outlet fits, any appliance (CreditManager, PricingEngine, event emitter) works with any store. This is the dependency inversion principle in action: high-level modules depend on abstractions, not concrete implementations.
The ABC is split into a core surface and an optional-capability surface. The core surface -- balance and atomic charging (get_balance, add_credits, deduct_with_allowance, refund_credits), the lease lifecycle (create_lease, settle_lease, release_lease, renew_lease, get_available), pricing config (get_active_pricing, set_active_pricing, get_pricing_history, get_pricing_config, activate_pricing), plans (get_user_plan, set_user_plan, check_allowance, increment_usage_window), spend-cap checks (check_spend_cap), and expiry sweeps (sweep_expired_credits) -- is @abstractmethod, so Python enforces it at instantiation time. The optional groups -- analytics (spend_by_user, spend_by_model, top_users, daily_spend, aggregate_stats), transaction listing (list_user_transactions), and shared team pools (create_team, get_team_balance, add_team_member, get_team_members, deduct_team) -- are concrete on the ABC with a default body that raises CapabilityNotSupportedError. A minimal custom store does not need to implement (or even think about) those groups at all; it only pays for them if it opts in by overriding the method.
This split matters for adopters: implementing "pluggable storage" against bursar means implementing roughly 20 core methods, not the full ~35-method surface. If a caller reaches for a capability your store does not support, they get a clear, typed CapabilityNotSupportedError instead of a confusing AttributeError or silently wrong data.
What we will do in this section: implement a minimal MyCustomStore that satisfies only the core ABC, wire it to a CreditManager, run it through a charge and a lease, and then show what happens when we call an optional capability it never implemented.
Implement the core ABC
Below is a minimal MyCustomStore -- dict-backed, no persistence, no concurrency guarantees beyond a single process lock. It implements exactly the core abstract methods and nothing else. Because none of the optional-capability methods are overridden, calling one of them (e.g. spend_by_user) falls through to the ABC's default body and raises CapabilityNotSupportedError -- which we will see below.
What we will do in this section: walk through the imports and the class definition, grouped by the same sections used in interface/base.py.
import uuid
import threading
from decimal import Decimal
from bursar.interface.base import CreditStore
from bursar.interface.models import (
BalanceResult, AddCreditsResult, DeductionResult, LeaseResult,
ReleaseResult, AvailableResult, RefundResult, GetUserPlanResult,
SetUserPlanResult, AllowanceResult, CapCheckResult, SetupResult,
SweepResult,
)
class MyCustomStore(CreditStore):
"""Minimal custom store -- dict-backed, single-process lock, no persistence."""
def __init__(self):
self._balances: dict[str, Decimal] = {}
# lease_id -> [user_id, amount, status] ("active" | "settled" | "released")
self._leases: dict[str, list] = {}
self._lock = threading.Lock()
def _held(self, user_id: str) -> Decimal:
return sum(
(amt for uid, amt, status in self._leases.values() if uid == user_id and status == "active"),
Decimal(0),
)
# ── Setup, balance, atomic charge, refund ────────────────────────────
def setup(self, database_url=None) -> SetupResult:
return SetupResult()
def get_balance(self, user_id: str) -> BalanceResult:
return BalanceResult(user_id=user_id, balance=self._balances.get(user_id, Decimal(0)))
def add_credits(self, user_id: str, amount, type: str = "adjustment",
metadata=None, expires_at=None) -> AddCreditsResult:
with self._lock:
amount = Decimal(amount)
new_balance = self._balances.get(user_id, Decimal(0)) + amount
self._balances[user_id] = new_balance
return AddCreditsResult(transaction_id=str(uuid.uuid4()), user_id=user_id,
amount=amount, new_balance=new_balance)
def deduct_with_allowance(self, user_id: str, amount, *, idempotency_key=None,
min_balance=Decimal(0), model=None, metadata=None,
skip_allowance=False, period_start=None) -> DeductionResult:
with self._lock:
amount = Decimal(amount)
balance = self._balances.get(user_id, Decimal(0))
if balance - amount < min_balance:
return DeductionResult(transaction_id="", user_id=user_id, amount=Decimal(0),
balance_after=balance, error="insufficient_credits")
balance -= amount
self._balances[user_id] = balance
return DeductionResult(transaction_id=str(uuid.uuid4()), user_id=user_id,
amount=amount, balance_after=balance)
def refund_credits(self, transaction_id: str, amount=None, reason=None, metadata=None) -> RefundResult:
return RefundResult(refund_transaction_id=str(uuid.uuid4()),
original_transaction_id=transaction_id, user_id="",
amount=amount or Decimal(0), new_balance=Decimal(0))
# ── Lease lifecycle (the only admission gate) ────────────────────────
# This toy store never expires a lease's TTL -- a real implementation
# would track expires_at and reject settle/renew on an elapsed lease.
def create_lease(self, user_id: str, amount, operation_type: str, *, billing_mode="strict",
floor=Decimal(0), max_concurrent=None, ttl_seconds=600, model=None,
overdraft_floor=None, metadata=None, period_start=None) -> LeaseResult:
with self._lock:
amount = Decimal(amount)
balance = self._balances.get(user_id, Decimal(0))
held = self._held(user_id)
available = balance - held
if available - amount < floor:
return LeaseResult(lease_id="", user_id=user_id, error="insufficient_credits")
lease_id = str(uuid.uuid4())
self._leases[lease_id] = [user_id, amount, "active"]
return LeaseResult(lease_id=lease_id, user_id=user_id, amount=amount,
available=available - amount, reserved_total=held + amount,
billing_mode=billing_mode)
def settle_lease(self, user_id: str, lease_id: str, amount, *, idempotency_key=None,
min_balance=Decimal(0), model=None, metadata=None,
skip_allowance=False, period_start=None) -> DeductionResult:
with self._lock:
entry = self._leases.get(lease_id)
if entry is None or entry[0] != user_id or entry[2] != "active":
return DeductionResult(transaction_id="", user_id=user_id, amount=Decimal(0),
balance_after=self._balances.get(user_id, Decimal(0)),
error="lease_not_found")
amount = Decimal(amount)
entry[2] = "settled"
balance = self._balances.get(user_id, Decimal(0)) - amount
self._balances[user_id] = balance
return DeductionResult(transaction_id=str(uuid.uuid4()), user_id=user_id,
amount=amount, balance_after=balance)
def release_lease(self, user_id: str, lease_id: str) -> ReleaseResult:
with self._lock:
entry = self._leases.get(lease_id)
if entry is None or entry[0] != user_id:
return ReleaseResult(lease_id=lease_id, user_id=user_id, released=False, reason="lease_not_found")
if entry[2] != "active":
return ReleaseResult(lease_id=lease_id, user_id=user_id, released=False, reason="already_finalized")
entry[2] = "released"
return ReleaseResult(lease_id=lease_id, user_id=user_id, released=True, reason="released")
def renew_lease(self, user_id: str, lease_id: str, ttl_seconds: int) -> LeaseResult:
entry = self._leases.get(lease_id)
if entry is None or entry[0] != user_id or entry[2] != "active":
return LeaseResult(lease_id=lease_id, user_id=user_id, error="lease_not_found")
return LeaseResult(lease_id=lease_id, user_id=user_id, amount=entry[1])
def get_available(self, user_id: str) -> AvailableResult:
balance = self._balances.get(user_id, Decimal(0))
held = self._held(user_id)
return AvailableResult(user_id=user_id, balance=balance, reserved=held, available=balance - held)
# ── Pricing config -- delegated to CreditManager/PricingEngine here ──
def get_active_pricing(self):
return None
def set_active_pricing(self, config, label=None) -> str:
return str(uuid.uuid4())
def get_pricing_history(self):
return []
def get_pricing_config(self, version: int):
return None
def activate_pricing(self, version: int) -> str:
return ""
# ── Plans -- this store has no plan catalog, so every user is planless ─
def get_user_plan(self, user_id: str) -> GetUserPlanResult:
return GetUserPlanResult(user_id=user_id)
def set_user_plan(self, user_id: str, plan_id: str) -> SetUserPlanResult:
return SetUserPlanResult(user_id=user_id, plan_id=plan_id)
def check_allowance(self, user_id: str, period_start=None) -> AllowanceResult:
return AllowanceResult(plan_id="", allowance_remaining=Decimal(0), period_start="", period_end="")
def increment_usage_window(self, user_id: str, plan_id: str, amount) -> None:
pass
# ── Spend caps and expiry -- no caps or expiry tracking in this toy ──
def check_spend_cap(self, user_id: str, model=None, amount=None) -> CapCheckResult:
return CapCheckResult()
def sweep_expired_credits(self, dry_run: bool = False) -> SweepResult:
return SweepResult()
# Instantiate our store. Python checks that all @abstractmethod-decorated
# methods are implemented at instantiation time -- since we implemented
# exactly the core surface (and none of the optional capabilities), this
# succeeds without needing a single team/analytics/transaction-listing method.
custom_store = MyCustomStore()
print("MyCustomStore implements the CreditStore core ABC.")
Use with CreditManager
Implementing the core ABC is enough to plug straight into CreditManager. The manager does not care whether the store persists to Postgres, Redis, DynamoDB, or a Python dictionary -- it only relies on the core contract. Below we add credits, then run both charging patterns from earlier notebooks (an atomic charge via deduct_with_allowance, Notebook 03, and a reserve → settle lease, Notebook 06) against our custom store.
What we will do in this section: create a CreditManager backed by our custom store, add credits, run an atomic charge and a reserve/settle lease, then confirm that this minimal store still works correctly for every core operation.
from bursar.manager import CreditManager
# Create a CreditManager using our custom store as the backend. It needs no
# pricing engine here since we charge raw Decimal amounts directly.
manager = CreditManager(custom_store)
user = str(uuid.uuid4())
manager.add_credits(user, Decimal("10_000"))
print(f" After adding 10,000 credits, balance = {manager.get_balance(user).balance}")
# Atomic charge (no reservation needed for a single-shot deduction). This
# calls the store directly, same as the deduct_with_allowance() pattern
# from Notebook 03 -- no pricing engine required for a raw amount.
ded = custom_store.deduct_with_allowance(user, Decimal("1_500"))
print(f" Charged 1,500 directly, balance = {ded.balance_after}")
# Lease lifecycle: reserve the worst-case hold, then settle the actual cost.
lease = manager.reserve(user, Decimal("1_000"))
print(f" Reserved {lease.amount}, available now {lease.available}")
settled = manager.settle(user, lease.lease_id, Decimal("600"))
print(f" Settled {settled.amount}, balance = {settled.balance_after}") # unused 400 hold freed
Optional capabilities raise a clear, typed error
Our store never overrode any of the optional-capability groups. Calling one -- for example spend_by_user, an analytics method -- falls through to the ABC's default body, which raises CapabilityNotSupportedError rather than an AttributeError or a silently empty/wrong result. This is the practical benefit of the WS8 core/optional split: callers can distinguish "this store doesn't support analytics" from "this store is broken," and a minimal custom store never has to implement (or stub out) a capability its application does not need.
What we will do in this section: call spend_by_user on our custom store and catch the resulting CapabilityNotSupportedError.
from datetime import datetime, timezone
from bursar.interface.base import CapabilityNotSupportedError
try:
custom_store.spend_by_user(datetime.now(timezone.utc), datetime.now(timezone.utc))
except CapabilityNotSupportedError as e:
print(f" spend_by_user raised CapabilityNotSupportedError: {e}")