Pomiar użycia, limity i integracja z rozliczeniami
Śledź użycie dla poszczególnych dzierżawców, egzekwuj limity planów i emituj zdarzenia rozliczeniowe na potrzeby rozliczeń SaaS opartych na użyciu.
Pomiar użycia, limity i integracja z rozliczeniami to bezpłatna lekcja FastAPI Backend Development Bootcamp na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej FastAPI Backend Development Bootcamp, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.
Części tej lekcji nie zostały jeszcze przetłumaczone i są wyświetlane po angielsku.
Why Metering Matters in SaaS
In a multi-tenant SaaS, every tenant shares the same FastAPI app but pays based on what they actually use. To support plans like Free / Pro / Enterprise you need three cooperating subsystems:
- Metering — count consumed resources (API calls, tokens, rows, GB).
- Quotas — block or throttle once a tenant exceeds its plan limit.
- Billing hooks — emit events that downstream billing (Stripe, etc.) turns into invoices.
The hard part is doing this per tenant, accurately, and without adding latency to every request. We will build each piece in this lesson.
Modeling Plans and Quotas
Start by modeling what each plan allows. A plan maps a metric (what we count) to a limit and a period (when the counter resets). Keeping this as plain data makes it easy to load from a DB or config.
Below, a metered plan tracks API requests and AI tokens with monthly limits. None means unlimited.
from dataclasses import dataclass
@dataclass(frozen=True)
class Quota:
metric: str
limit: int | None # None = unlimited
period: str # 'day' or 'month'
PLANS = {
"free": [Quota("api_calls", 1_000, "month"), Quota("ai_tokens", 50_000, "month")],
"pro": [Quota("api_calls", 100_000, "month"), Quota("ai_tokens", 5_000_000, "month")],
"enterprise": [Quota("api_calls", None, "month"), Quota("ai_tokens", None, "month")],
}
def limit_for(plan: str, metric: str) -> int | None:
for q in PLANS[plan]:
if q.metric == metric:
return q.limit
return 0 # metric not allowed on this plan
print(limit_for("free", "api_calls")) # 1000
print(limit_for("enterprise", "ai_tokens")) # None (unlimited)
print(limit_for("free", "unknown")) # 0An Atomic Usage Counter
The core of metering is a counter you can increment atomically per (tenant, metric, period). In production this is usually Redis (INCRBY + EXPIRE) or a Postgres UPSERT with an atomic +=.
Here is the concept modeled in pure Python so you can see the key and the increment logic. The period key (e.g. 2026-06) is what makes the counter reset every month automatically.
from collections import defaultdict
from datetime import datetime, timezone
usage = defaultdict(int)
def period_key(period: str, now: datetime) -> str:
if period == "month":
return now.strftime("%Y-%m")
return now.strftime("%Y-%m-%d")
def incr(tenant: str, metric: str, amount: int, period: str, now: datetime) -> int:
key = (tenant, metric, period_key(period, now))
usage[key] += amount
return usage[key]
now = datetime(2026, 6, 10, tzinfo=timezone.utc)
print(incr("acme", "api_calls", 1, "month", now)) # 1
print(incr("acme", "api_calls", 1, "month", now)) # 2
print(incr("acme", "ai_tokens", 1200, "month", now)) # 1200Enforcing a Quota
Quota enforcement combines the plan limit with the current counter. The decision is simple but the order matters: you usually want to check-then-increment so that a request that would exceed the limit is rejected before doing the work.
For atomicity in real systems you increment first, then compare against the limit, and if you went over, you reject and optionally decrement (or simply let the counter sit over-limit, blocking further calls).
def check_quota(current: int, amount: int, limit: int | None) -> bool:
"""Return True if `amount` more units are allowed."""
if limit is None:
return True # unlimited
return current + amount <= limit
print(check_quota(995, 1, 1000)) # True (996 <= 1000)
print(check_quota(1000, 1, 1000)) # False (1001 > 1000)
print(check_quota(9_000_000, 1, None)) # True (unlimited)Wiring It Into FastAPI with a Dependency
In FastAPI the clean place to enforce quotas is a dependency. It resolves the tenant (from auth/JWT), looks up the plan, checks the quota, and raises HTTPException(429) when the limit is hit.
Returning 402 Payment Required is also common for hard plan caps, while 429 Too Many Requests fits rate-style limits. This is framework code, so it is not standalone-runnable.
from fastapi import Depends, HTTPException, Request
async def enforce_quota(metric: str, cost: int = 1):
async def _dep(request: Request):
tenant = request.state.tenant_id
plan = await get_plan(tenant)
limit = limit_for(plan, metric)
current = await store.get(tenant, metric, "month")
if not check_quota(current, cost, limit):
raise HTTPException(
status_code=429,
detail=f"Quota exceeded for {metric}",
headers={"X-Quota-Limit": str(limit)},
)
await store.incr(tenant, metric, cost, "month")
return current + cost
return _dep
@app.post("/v1/complete")
async def complete(used=Depends(enforce_quota("ai_tokens", cost=500))):
return {"ok": True, "tokens_used_this_month": used}Metering AFTER the Work for Variable Costs
Some costs are not known up front. AI token usage, bytes processed, or rows returned are only known after the handler runs. For these, do a cheap pre-check (is the tenant already over?) and record the real cost afterward.
A reliable pattern is a context manager that admits the request, runs the body, then reports actual usage — even on exceptions, so you don't lose metering data.
from contextlib import contextmanager
@contextmanager
def meter(record, tenant, metric):
actual = {"amount": 0}
try:
yield actual # caller sets actual['amount']
finally:
if actual["amount"]:
record(tenant, metric, actual["amount"])
log = []
with meter(lambda t, m, a: log.append((t, m, a)), "acme", "ai_tokens") as use:
# ... do real work, then we learn the true cost ...
use["amount"] = 1342
print(log) # [('acme', 'ai_tokens', 1342)]Soft Limits, Hard Limits and Overage
Real billing rarely flips from 'works' to '403' at exactly the limit. Plans usually define:
- Soft limit — allow but warn (email,
X-Quota-Warningheader) around ~80%. - Hard limit — block once exceeded (Free plans).
- Overage — keep serving but bill each extra unit at a metered rate (Pro/Enterprise).
Encode this as a policy so the enforcement code stays declarative.
def classify(current: int, limit: int | None, overage_allowed: bool) -> str:
if limit is None:
return "ok"
if current < 0.8 * limit:
return "ok"
if current < limit:
return "warn"
return "overage" if overage_allowed else "blocked"
print(classify(700, 1000, False)) # ok
print(classify(850, 1000, False)) # warn
print(classify(1200, 1000, True)) # overage
print(classify(1200, 1000, False)) # blockedEmitting Billing Events
Metering produces a stream of usage events. Billing systems (Stripe Billing Meters, Orb, Metronome) consume these to compute invoices. The golden rules for billing events:
- Include an idempotency key so retries don't double-charge.
- Carry a timestamp so events land in the correct billing period.
- Make them immutable — append-only, never edit.
Build the event payload as a plain, serializable record.
import uuid, json
from datetime import datetime, timezone
def billing_event(tenant: str, metric: str, qty: int, ts: datetime, request_id: str) -> dict:
return {
"id": str(uuid.uuid4()),
"idempotency_key": f"{tenant}:{metric}:{request_id}",
"tenant_id": tenant,
"metric": metric,
"quantity": qty,
"occurred_at": ts.isoformat(),
}
evt = billing_event("acme", "ai_tokens", 1342,
datetime(2026, 6, 10, tzinfo=timezone.utc), "req_9f3")
print(json.dumps(evt, indent=2))Decoupling with a Background Queue
Never call your billing provider synchronously inside the request path — a slow or down Stripe API would break your API. Instead, write the usage event to a durable buffer and flush it asynchronously.
FastAPI's BackgroundTasks works for low volume; for real scale use an outbox table, Redis stream, or a worker (Celery/ARQ). The handler stays fast; a separate consumer ships events to billing with retries and idempotency.
from fastapi import BackgroundTasks
@app.post("/v1/complete")
async def complete(bg: BackgroundTasks, request: Request):
tenant = request.state.tenant_id
result, tokens = await run_completion(request)
await store.incr(tenant, "ai_tokens", tokens, "month")
evt = billing_event(tenant, "ai_tokens", tokens,
datetime.now(timezone.utc), request.state.request_id)
bg.add_task(emit_to_billing, evt) # fire-and-forget, off the hot path
return resultReconciliation: Trust but Verify
Counters drift. A crashed worker, a lost event, or a Redis flush can desync your metering from reality. Robust SaaS systems treat the append-only event log as the source of truth and periodically reconcile the fast counters against it.
A nightly job re-aggregates raw events per tenant/metric/period and corrects the cached counter. This also lets you regenerate billing if a provider rejected a batch.
from collections import defaultdict
def reconcile(events: list[dict]) -> dict:
"""Rebuild authoritative totals from the immutable event log."""
totals = defaultdict(int)
seen = set()
for e in events:
if e["idempotency_key"] in seen:
continue # dedupe replays
seen.add(e["idempotency_key"])
period = e["occurred_at"][:7] # YYYY-MM
totals[(e["tenant_id"], e["metric"], period)] += e["quantity"]
return dict(totals)
events = [
{"idempotency_key": "acme:t:1", "tenant_id": "acme", "metric": "t", "quantity": 100, "occurred_at": "2026-06-01T10:00:00"},
{"idempotency_key": "acme:t:1", "tenant_id": "acme", "metric": "t", "quantity": 100, "occurred_at": "2026-06-01T10:00:00"},
{"idempotency_key": "acme:t:2", "tenant_id": "acme", "metric": "t", "quantity": 50, "occurred_at": "2026-06-02T09:00:00"},
]
print(reconcile(events)) # {('acme','t','2026-06'): 150}Tenant Isolation and Hot Keys
A few operational rules keep per-tenant metering correct and fast:
- Always scope the counter key by tenant_id — never share a global counter, or one noisy tenant blocks others.
- Watch for hot keys: a huge tenant hammering one Redis key can become a bottleneck; shard the key (
tenant:metric:shard) and sum on read. - Make limit checks fail-open or fail-closed deliberately — if the counter store is down, decide whether to allow (availability) or block (revenue protection).
- Reset boundaries must respect the tenant's billing-cycle timezone, not just UTC month.
Quick Check
Test your understanding of the metering decision flow.
Recap
You built a complete per-tenant metering and billing pipeline:
- Plans & quotas as data: metric to limit to reset period, with
Nonefor unlimited. - Atomic counters keyed by
(tenant, metric, period)so limits reset automatically. - Enforcement via a FastAPI dependency returning
429/402; check-then-increment for fixed costs, meter-after for variable costs. - Policies for soft limits, hard limits, and overage instead of a single hard cutoff.
- Billing events that are immutable, timestamped, and idempotent, shipped asynchronously off the request path.
- Reconciliation from an append-only event log as the source of truth, plus tenant-isolation and hot-key care.
With these pieces you can support usage-based pricing safely without slowing down your API.
Ucz się FastAPI Backend Development Bootcamp dzięki korepetycjom AI — za darmo
Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.
- Kursy
- 21
- Lekcje
- 84
Często zadawane pytania
Czy lekcja „Pomiar użycia, limity i integracja z rozliczeniami” jest bezpłatna?
Tak — pełny tekst „Pomiar użycia, limity i integracja z rozliczeniami” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu FastAPI Backend Development Bootcamp, przejdź na CoddyKit PRO. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.
Co nauczysz się w „Pomiar użycia, limity i integracja z rozliczeniami”?
Śledź użycie dla poszczególnych dzierżawców, egzekwuj limity planów i emituj zdarzenia rozliczeniowe na potrzeby rozliczeń SaaS opartych na użyciu. Ćwiczysz FastAPI Backend Development Bootcamp z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć FastAPI Backend Development Bootcamp?
Nie wymagamy żadnego doświadczenia. FastAPI Backend Development Bootcamp w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „Pomiar użycia, limity i integracja z rozliczeniami”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji FastAPI Backend Development Bootcamp?
Tak. Każda lekcja FastAPI Backend Development Bootcamp zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Strategie izolacji dzierżawców i związane z nimi kompromisy
- Rozpoznawanie kontekstu dzierżawcy i middleware
- Bezpieczeństwo na poziomie wierszy i partycjonowanie danych
- Pomiar użycia, limity i integracja z rozliczeniami