:::info Executable tutorial
This page is generated from a tested Jupyter notebook. Open it in Google Colab or view the source notebook.
:::
Manage credit priority and expiry
Credit buckets let promotional and purchased value follow different consumption and expiry policies. This tutorial grants expiring promotional credits, consumes them before purchased credits, runs the expiry sweep, and verifies every change in the ledger.
Learning objectives
After completing this tutorial, you can:
- Configure and inspect bucket priority
- Grant credits with an expiry timestamp
- Preview and execute an expiry sweep
- Confirm that expiry posts an accounting entry
Prerequisites
- Complete the credit lifecycle tutorial
- Start the notebook server from
samples/python/notebooks/
Setup
Same sandbox: a temporary Postgres store, USER_ADA on pro with 100 purchased credits in the purchased bucket.
from decimal import Decimal
from datetime import UTC, datetime, timedelta
from time import sleep
from bursar.metrics import UsageMetrics
from shared import cleanup, base_config, publish_config, start_postgres_store, USER_ADA
def completion(output=Decimal(500)):
return UsageMetrics(
operation="completion",
measures={
"input_tokens": Decimal(1000),
"output_tokens": output,
"cache_read_tokens": Decimal(200),
},
dimensions={"model": "gpt-4o"},
)
store, pgdata = start_postgres_store()
bursar = publish_config(store, base_config())
bursar.accounts.on_account_created(USER_ADA, "signup")
bursar.credits.set_user_plan(USER_ADA, "pro")
bursar.credits.add_credits(USER_ADA, Decimal("100.00"), entry_type="purchase", idempotency_key="buy")
print("ready")
Grants with a shelf life
A 20.00 promotional grant is issued with expires_at a few seconds in the future, landing in the promotional tier while the purchase sits in purchased. Balance is 120.000000.
The bucket view reports expires=False for both tiers: that flag mirrors the bucket definition in the published config (whether the bucket is expiry-enabled), and the standard config leaves it off. Expiry here is per lot, set at grant time with expires_at — the sweep honors it regardless of the flag.
grant = bursar.credits.add_credits(
USER_ADA,
Decimal("20.00"),
entry_type="grant",
bucket="promotional",
expires_at=datetime.now(UTC) + timedelta(seconds=5),
idempotency_key="trial-grant",
)
print("grant entry:", grant.entry_id[:12], "| new balance:", grant.new_balance)
for b in bursar.credits.get_bucket_balances(USER_ADA).buckets:
print(f"bucket={b.bucket_key!r} expires={b.expires} balance={b.balance}")
While it lives
Promotional credits sit at priority 1, so usage spends from them first. A small chat costs 0.000008, all of it from the promotional lot.
r = bursar.credits.deduct(USER_ADA, completion(), idempotency_key="promo-chat")
print("amount:", r.amount, "| breakdown:", r.bucket_breakdown)
promo = bursar.credits.get_bucket_balances(USER_ADA).buckets[0]
print("promotional remaining:", promo.balance)
Expiry and the sweep
After the expiry moment passes, the balance still shows the credits — expiry is only acted on by a sweep, and the sweep can be dry-run first. Expect the dry run to report 1 lot / 19.999992 credits (20.00 minus the chat), then the real sweep to remove it, leaving exactly the purchased 100.000000.
sleep(6)
print("balance before sweep:", bursar.credits.get_balance(USER_ADA).balance)
dry = bursar.credits.sweep_expired_credits(dry_run=True)
print("dry run:", dry.expired_count, "lots,", dry.expired_amount, "credits, by bucket:", dry.expired_by_bucket)
real = bursar.credits.sweep_expired_credits(dry_run=False)
print("real sweep:", real.expired_count, "lots,", real.expired_amount, "credits, by bucket:", real.expired_by_bucket)
print("balance after sweep:", bursar.credits.get_balance(USER_ADA).balance)
Past expiry is rejected
Grants are validated on the way in as well: an expires_at in the past is refused by the store with StoreError — the API refuses to create a lot that is already dead.
try:
bursar.credits.add_credits(
USER_ADA,
Decimal("10.00"),
entry_type="grant",
bucket="promotional",
expires_at=datetime.now(UTC) - timedelta(seconds=1),
idempotency_key="trial-grant-expired",
)
except Exception as exc:
print("past expiry:", type(exc).__name__, getattr(exc, "code", None), "|", str(exc)[:60])
Tiers in the ledger
The ledger tells the whole story: the 100.00 purchase, the 20.00 grant, the small usage that ate into it, and the negative expiry entry that retired the rest.
for e in bursar.credits.list_ledger_entries(USER_ADA, limit=6).items:
print(f" {e.entry_type:8s} {e.amount:>12} {e.created_at[:19]}")