Architecting a Multi-Tenant SaaS on Gemini API — Tenant Isolation, Usage Metering, and Runaway Cost Defense in Production

Decision 2: tenant authentication and the request context

Every Gemini call should carry a "which tenant, which user, which plan" context. If you push this into a TenantContext value object at the entry point and thread it through the call stack, your billing, audit, and budget code all fall out naturally later.

# tenant_context.py
from dataclasses import dataclass
from typing import Literal
 
PlanTier = Literal["free", "starter", "pro", "enterprise"]
 
@dataclass(frozen=True)
class TenantContext:
    """Per-request tenant info, created in the auth middleware
    and threaded down to the Gemini call sites."""
 
    tenant_id: str            # e.g. "tnt_01HX..."
    user_id: str              # e.g. "usr_01HX..."
    plan: PlanTier            # billing plan
    monthly_token_budget: int # remaining token budget for this billing month
    request_id: str           # unique trace id
 
    @property
    def has_budget(self) -> bool:
        return self.monthly_token_budget > 0

Expose this through Depends in FastAPI, g in Flask, or c.set() in Hono so every endpoint can read it. The unbreakable rule is: no Gemini call may bypass authentication. A common mistake is "this internal cron job doesn't need auth" — that one cron job ends up emitting cost that you can't attribute to any tenant.

Decision 3: per-tenant rate limiting with token buckets

Gemini's own quota lives at the project level, so you need an app-side rate limiter that guarantees Tenant A's success rate and latency don't degrade just because Tenant B is misbehaving.

I like to implement this with a Redis token bucket: it gives sub-millisecond decisions, holds up under multi-instance API servers, and a Lua script lets us refill and consume atomically.

# tenant_rate_limiter.py
import time
import redis.asyncio as redis
 
# Lua: atomically refill, read, and decrement the bucket.
# Returns remaining tokens, or -1 to signal "deny".
LUA_TOKEN_BUCKET = """
local key = KEYS[1]
local rate_per_sec = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local cost = tonumber(ARGV[4])
 
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now
 
local elapsed = math.max(0, now - last_refill)
tokens = math.min(capacity, tokens + elapsed * rate_per_sec)
 
local allowed = tokens >= cost
if allowed then
  tokens = tokens - cost
end
 
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return allowed and tokens or -1
"""
 
class TenantRateLimiter:
    # plan -> (rate tokens/sec, burst capacity)
    PLAN_LIMITS = {
        "free":       (1.0,   10),
        "starter":    (5.0,   50),
        "pro":        (20.0,  200),
        "enterprise": (100.0, 1000),
    }
 
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self._sha: str | None = None
 
    async def _ensure_loaded(self) -> str:
        if self._sha is None:
            self._sha = await self.redis.script_load(LUA_TOKEN_BUCKET)
        return self._sha
 
    async def try_acquire(self, ctx, cost: int = 1) -> bool:
        """Should this tenant be allowed to send one more request right now?
        Pass a non-1 cost when you want heavy requests to consume more tokens."""
        rate, capacity = self.PLAN_LIMITS[ctx.plan]
        sha = await self._ensure_loaded()
        key = f"ratelimit:tenant:{ctx.tenant_id}"
        result = await self.redis.evalsha(
            sha, 1, key, rate, capacity, time.time(), cost
        )
        return result != -1

A note on the clock: now is passed from Python's time.time(), so clock skew between API servers shows up in the refill amount. If that bothers you, you can call redis.call('TIME') inside the Lua script instead (try it on your Redis version first).

Two things matter here. First, the limits must differ by plan. SaaS plans usually charge more for higher rate limits, but if you store the plan only in Stripe and never read it from the rate limiter, "I'm on Pro but my throughput is the same as Free" is a complaint you will absolutely receive.

Second, decide the deny-path UX in advance. When try_acquire returns False, you can (a) immediately return 429, (b) queue briefly and serve later, or (c) return 429 with a Retry-After header. For chat UIs I prefer (c). For webhook-style consumers, (a) is more honest because the caller knows how to retry.

Decision 4: request-level token metering for billing

Equal in importance to rate limiting is recording exactly how much each tenant used, at request granularity, so your monthly invoices are defensible. A SaaS that can't answer "how much did Tenant X spend last month?" loses trust quickly.

Gemini returns usage_metadata on every response with prompt_token_count, candidates_token_count, and total_token_count. Capture them every time.

# usage_recorder.py
import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone
 
from google import genai
from google.genai import types
 
# Per-model pricing in USD per 1M tokens (sample values, April 2026)
MODEL_PRICING = {
    "gemini-2.5-pro":         {"input": 1.25, "output": 10.00},
    "gemini-2.5-flash":       {"input": 0.10, "output": 0.40},
    "gemini-2.5-flash-lite":  {"input": 0.05, "output": 0.20},
}
 
@dataclass
class UsageRecord:
    tenant_id: str
    user_id: str
    request_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    created_at: datetime
    success: bool
    error_code: str | None = None
 
def _calc_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    p = MODEL_PRICING.get(model, {"input": 1.0, "output": 5.0})
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000
 
class GeminiClientWithMetering:
    """Wraps a Gemini client so every call also persists a usage record.
    The caller just provides ctx and prompt."""
 
    def __init__(self, client: genai.Client, recorder, budget_guard):
        self.client = client
        self.recorder = recorder
        self.budget_guard = budget_guard
 
    async def generate(self, ctx, model: str, prompt: str) -> str:
        if not await self.budget_guard.allow(ctx):
            raise PermissionError(f"tenant {ctx.tenant_id} over monthly budget")
 
        record = UsageRecord(
            tenant_id=ctx.tenant_id,
            user_id=ctx.user_id,
            request_id=ctx.request_id,
            model=model,
            input_tokens=0,
            output_tokens=0,
            cost_usd=0.0,
            created_at=datetime.now(timezone.utc),
            success=False,
        )
 
        try:
            resp = await self.client.aio.models.generate_content(
                model=model,
                contents=prompt,
            )
            usage = resp.usage_metadata
            record.input_tokens = usage.prompt_token_count or 0
            # Thinking tokens are billed at the output rate but are NOT in
            # candidates_token_count. Add the separate field
            record.output_tokens = (
                (usage.candidates_token_count or 0)
                + (usage.thoughts_token_count or 0)
            )
            record.cost_usd = _calc_cost(
                model, record.input_tokens, record.output_tokens
            )
            record.success = True
            return resp.text
 
        except Exception as e:
            record.error_code = type(e).__name__
            raise
        finally:
            # Always persist — even on failure. Some failures still incur charges
            # (for example, a network drop during streaming output).
            await self.recorder.persist(record)

The non-obvious detail here is always calling recorder.persist() from the finally block. Even when an exception fires, you may have been charged for input tokens already (especially with streaming, where the network can drop mid-output). Implementations that "only record on success" reliably under-report by a few percent, and that's the few percent that turns into dispute calls at the end of the month.

A hole I found on review: thinking tokens, and failed calls

In my first version of this section the output side was counted from candidates_token_count alone. What stopped me was rereading one line of the official docs: with thinking on, response pricing is the sum of output tokens and thinking tokens. candidates_token_count holds only the visible answer; the tokens spent thinking arrive separately in thoughts_token_count.

Take a call with 1,000 input tokens, a 200-token visible answer and 3,000 thinking tokens, priced with the Flash row of the table above:

That is 16x on the output side and about 7.7x on the whole call. If your records silently drop that difference, both the customer invoice and the budget guard drift below real spend, and because thinking volume swings from prompt to prompt you can't correct for it with an average. Take the number you bill from what is charged, not from what is visible — that's the line I drew. The code above now adds thoughts_token_count; the or 0 is there because the field can be None depending on the SDK version and the model.

The gemini-2.5-* IDs in the code are examples. Google's changelog in September 2026 limited the 2.5 family to recent users, so if you're starting fresh, swap in a current Flash model ID and load prices from the official pricing page rather than hard-coding them.

One more gap: a failed call recorded in the finally block is stored with zero output tokens, even when Google may have charged for it. Code can't close that difference. Once a month, compare the sum of usage_records against your Cloud Billing export (BigQuery) and look at the size of the gap. While it's small, leave it alone; past some percentage you set, suspect dropped records.

A reasonable Postgres schema for usage_records looks like this:

-- migrations/001_usage_records.sql
CREATE TABLE usage_records (
    id              BIGSERIAL PRIMARY KEY,
    tenant_id       TEXT NOT NULL,
    user_id         TEXT NOT NULL,
    request_id      TEXT NOT NULL UNIQUE,  -- prevents double-charging on retry
    model           TEXT NOT NULL,
    input_tokens    INTEGER NOT NULL,
    output_tokens   INTEGER NOT NULL,
    cost_usd        NUMERIC(10, 6) NOT NULL,
    success         BOOLEAN NOT NULL,
    error_code      TEXT,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
 
CREATE INDEX idx_usage_tenant_created
    ON usage_records (tenant_id, created_at DESC);
 
-- For monthly billing rollups
CREATE MATERIALIZED VIEW usage_monthly AS
SELECT
    tenant_id,
    date_trunc('month', created_at) AS month,
    SUM(input_tokens) AS input_tokens,
    SUM(output_tokens) AS output_tokens,
    SUM(cost_usd) AS cost_usd,
    COUNT(*) FILTER (WHERE success) AS ok_count,
    COUNT(*) FILTER (WHERE NOT success) AS error_count
FROM usage_records
GROUP BY tenant_id, date_trunc('month', created_at);

The UNIQUE constraint on request_id is the unsung hero of this schema. If a retry fires the same logical request twice and both INSERTs succeed, the customer gets billed twice. Generate request_id once at the request entry point and carry it all the way through.

If you start to feel write contention as you scale past ~10K requests/day, move the writes to Cloud Pub/Sub or Kafka and let a downstream consumer load BigQuery. Reads (rollups) will keep working off the same source of truth.

Decision 5: monthly budget guard and circuit breaker

Rate limiting protects you from short-term runaways. A budget guard protects you from medium-term ones. A tenant sending one request per second isn't fast — but if they keep that up for 24 hours a day, the bill at the end of the month is real.

I run two layers of defense.

# budget_guard.py
import asyncio
from datetime import datetime, timezone
 
import asyncpg
 
class TenantBudgetGuard:
    """Keeps a per-tenant 'spend so far this month' number cached briefly,
    and warns at 80% and blocks at 105% of plan budget."""
 
    PLAN_BUDGETS = {
        "free":       0.50,
        "starter":    20.0,
        "pro":        200.0,
        "enterprise": 2000.0,  # hard cap; negotiated up-front per contract
    }
 
    GRACE_RATIO = 1.05  # tolerate 5% so a lagging rollup doesn't reject borderline calls
 
    def __init__(self, pool: asyncpg.Pool, alerter):
        self.pool = pool
        self.alerter = alerter
        self._cache: dict[str, tuple[float, datetime]] = {}
        self._locks: dict[str, asyncio.Lock] = {}              # one lock per tenant
        self._alerted: set[tuple[str, str, str]] = set()       # (tenant, month, kind)
 
    async def _current_spend(self, tenant_id: str) -> float:
        # 60-second cache. We trade strict freshness for many fewer DB hits.
        lock = self._locks.setdefault(tenant_id, asyncio.Lock())
        async with lock:
            cached = self._cache.get(tenant_id)
            now = datetime.now(timezone.utc)
            if cached and (now - cached[1]).total_seconds() < 60:
                return cached[0]
 
            month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
            row = await self.pool.fetchrow(
                "SELECT COALESCE(SUM(cost_usd), 0) AS spend "
                "FROM usage_records "
                "WHERE tenant_id = $1 AND created_at >= $2",
                tenant_id, month_start,
            )
            spend = float(row["spend"])
            self._cache[tenant_id] = (spend, now)
            return spend
 
    async def _alert_once(self, kind: str, tenant_id: str, spend: float, budget: float):
        # One notification per tenant, month and kind ("warning" or "blocked")
        key = (tenant_id, datetime.now(timezone.utc).strftime("%Y-%m"), kind)
        if key in self._alerted:
            return
        self._alerted.add(key)
        await getattr(self.alerter, f"notify_{kind}")(tenant_id, spend, budget)
 
    async def allow(self, ctx) -> bool:
        budget = self.PLAN_BUDGETS[ctx.plan]
        spend = await self._current_spend(ctx.tenant_id)
 
        if spend >= budget * self.GRACE_RATIO:
            await self._alert_once("blocked", ctx.tenant_id, spend, budget)
            return False
 
        if spend >= budget * 0.80:
            await self._alert_once("warning", ctx.tenant_id, spend, budget)
 
        return True

The first version of this code had three holes I only saw on review.

The first was a lock shared by every tenant. One asyncio.Lock() was held across the DB rollup, so a slow rollup for a hot tenant made every unrelated tenant's allow() wait in line — the mechanism meant to separate tenants was serializing them. The code above keeps one lock per tenant. In a fake environment where the DB takes 0.05s to answer, ten tenants going through together took about 0.5s with the shared lock and about 0.05s with per-tenant locks.

The second was the warning condition. Notifying only while spend sits between 80% and 85% means that if spend jumps from 79% to 90% between cache refreshes, no warning ever fires; and while it stays under 85%, a notification goes out on every request. The condition is now "at or above 80%", with one notification per (tenant, month, kind). That set lives in process memory, so with several API servers you'll get one notification per server. Once you scale out, move it to Redis SET NX.

The third is how much overshoot the 60-second cache allows. The next part puts a number on it.

How far can the 60-second cache let you overshoot?

Computing _current_spend from Postgres on every request will cook your DB if any tenant gets hot. The 60-second cache is the realistic compromise — nearly real-time, but not exact — and I had never put a number on that "nearly".

The guard sees only spend confirmed up to 60 seconds ago. And since usage_records is written after a call finishes, requests still in flight are invisible too. So suppose the guard says "still room" and then, for the next 60 seconds, the tenant sends as many calls as the rate limiter from Decision 3 lets through. How far over can it go? I computed it from each plan's limits (burst capacity plus 60 seconds of refill). Each call is assumed to be Pro with 50K input and 4K output tokens at the sample prices (about $0.1025).

The 1,435% for free looks absurd, but it's what happens if free can call Pro. Restrict free and starter to Flash-class models and the same arithmetic gives about $0.46 and $2.31. I realized per-plan rate limits and per-plan model access have to be decided together when I laid this table out.

This is the worst alignment — the guard looking at a stale number just under the limit. It won't happen every time. Still, the precision of a monthly budget guard is set by "60 seconds x pass-through rate x cost per call", and that doesn't change.

A budget guard should watch not "how much was spent" but "how much can be spent in the next 60 seconds".

There are two ways to shrink that. Shorten the cache, which trades against DB load, or reserve the worst-case cost before the call. Here's the second. Keep a monthly estimated-spend counter in Redis, add the worst case before the call, roll it back and reject if it would exceed the budget, and settle with the actual cost afterwards.

# spend_ledger.py
from datetime import datetime, timezone
 
class SpendLedger:
    """Estimated spend for this month, kept in Redis. Reserve the worst case
    before a call, settle with the actual cost after."""
 
    def __init__(self, redis_client, db_spend, budgets: dict, grace: float = 1.05):
        self.redis = redis_client
        self.db_spend = db_spend      # async (tenant_id) -> confirmed spend this month (DB rollup)
        self.budgets = budgets
        self.grace = grace
 
    async def reserve(self, ctx, est_usd: float) -> str | None:
        """Returns the Redis key if reserved, or None if it would break the budget."""
        key = f"spend:{ctx.tenant_id}:{datetime.now(timezone.utc):%Y%m}"
        if not await self.redis.exists(key):
            # Seed once from the DB. NX means concurrent callers seed it only once
            await self.redis.set(key, await self.db_spend(ctx.tenant_id),
                                 nx=True, ex=40 * 86400)
        after = float(await self.redis.incrbyfloat(key, est_usd))
        if after > self.budgets[ctx.plan] * self.grace:
            await self.redis.incrbyfloat(key, -est_usd)   # give the reservation back
            return None
        return key
 
    async def settle(self, key: str, est_usd: float, actual_usd: float) -> None:
        """Apply the difference between estimate and actual. A failed call
        passes actual_usd=0, which releases the reservation."""
        await self.redis.incrbyfloat(key, actual_usd - est_usd)

On the calling side, use this flow in place of budget_guard.allow():

MAX_OUT = 4096   # max_output_tokens; the cap includes thinking tokens
 
p = MODEL_PRICING[model]
est = (len(prompt) * p["input"] + MAX_OUT * p["output"]) / 1_000_000
key = await ledger.reserve(ctx, est)
if key is None:
    raise PermissionError(f"tenant {ctx.tenant_id} over monthly budget")
try:
    resp = await client.aio.models.generate_content(
        model=model, contents=prompt,
        config=types.GenerateContentConfig(max_output_tokens=MAX_OUT),
    )
    ...  # put the actual cost in record.cost_usd
finally:
    await ledger.settle(key, est, record.cost_usd)   # 0 on failure releases it

The worst case uses character count for the input side and max_output_tokens for the output side. Character count tends to be at or above the token count in most languages, but that's a rule of thumb, not a guarantee. The official docs count thinking tokens inside the output cap, and a response that hits the cap is cut off, so it's a UX trade-off.

In my test, a tenant with $190 already confirmed in the DB was hit with 200 reservations of $0.1025 each; 195 were admitted and the total stopped at $209.99, just under 105% of budget. When all 195 were settled as failures, the counter went back to $190 (checked against a simple stand-in for Redis). A Redis counter drifts from the DB a little at a time, so pair it with a nightly overwrite from the usage_records rollup.

In addition, run a service-wide circuit breaker. When you detect sustained Gemini 5xx bursts or your Cloud Billing alarm fires, flip a Redis health key to down and short-circuit every Gemini call to a graceful error response. The detailed cost-spike playbook is in Defending Gemini API costs against runaway spend — it pairs nicely with the budget guard above.

Pitfalls I personally walked into

This isn't theory. Each of these cost me a real evening.

Pitfall 1: losing token counts in streaming responses

generate_content_stream() only puts usage_metadata on the final chunk. If you break early, or you accidentally drop the last chunk, your token counter is permanently zero. I noticed only when reconciling Cloud Billing against my DB at month-end. What happens on the server side after the client breaks the connection, and how that shows up in billing, is something my own logs can't confirm; a monthly reconciliation shows the size of that gap.

# Wrong
async for chunk in client.aio.models.generate_content_stream(...):
    yield chunk.text
    if some_condition:
        break  # last chunk discarded; usage_metadata gone
 
# Right
last_chunk = None
async for chunk in client.aio.models.generate_content_stream(...):
    last_chunk = chunk
    yield chunk.text
if last_chunk and last_chunk.usage_metadata:
    record.input_tokens = last_chunk.usage_metadata.prompt_token_count
    record.output_tokens = last_chunk.usage_metadata.candidates_token_count

Pitfall 2: forgetting to attach tenant_id to logs

A stack trace alone doesn't tell you which tenant is misbehaving. Make tenant_id, user_id, and request_id mandatory fields in your logging.extra. I use contextvars to bind them at the request boundary and a logging.Filter to inject them automatically, so every line is searchable by tenant.

Pitfall 3: leaking BYOK keys

When you graduate to Plan C (BYOK), storing tenant API keys in plaintext columns is the most common breach pattern in the industry. Use Cloud KMS at minimum, ideally HashiCorp Vault, and never let the encryption key live in your application repo. Even with that, breaches happen — pair the technical control with a customer-facing rule that says "the key you give us must be locked to read-only with a strict monthly budget cap from your side."

Pitfall 4: never deciding the retention period for usage_records

Leaving usage_records to grow forever ends in a multi-terabyte table that no one wants to migrate. Decide upfront: (a) keep 24 months hot then archive, (b) roll up into monthly summary tables, or (c) ETL into BigQuery. Local tax law may also dictate a minimum retention (seven years for Japan's accounting records, for example).

The minimum viable observability dashboard

Once the writes are in place, you have everything you need to graph. Five panels are usually enough:

• Per-tenant monthly spend (USD), with the top 10 tenants and a column showing how close they are to plan cap.
• Model mix (input vs output tokens) to understand the shape of your cost.
• Error rate per tenant and overall, broken down by 429 / 5xx / timeout.
• Average latency per tenant — drift here is your earliest signal of noisy-neighbor pain.
• Budget alerts fired, listing the tenants that crossed 80% or 100% and when.

I run this on Grafana over Postgres, but Looker Studio over BigQuery works equally well. So does a Google Sheet, honestly. Decide what "anomaly looks like" first, then build the smallest dashboard that reveals it.

Wiring it all together in FastAPI

To make the moving parts concrete, here is a minimal FastAPI service that authenticates a tenant, applies the rate limiter and budget guard, and serves a single chat endpoint backed by Gemini.

# app.py
from fastapi import FastAPI, Depends, HTTPException, Request
from contextvars import ContextVar
import uuid
 
from tenant_context import TenantContext
from tenant_rate_limiter import TenantRateLimiter
from usage_recorder import GeminiClientWithMetering
from budget_guard import TenantBudgetGuard
 
# Globally, you would build these once at startup.
limiter: TenantRateLimiter
gemini: GeminiClientWithMetering
guard: TenantBudgetGuard
 
# Bind tenant context to the current request for logging convenience.
current_ctx: ContextVar[TenantContext | None] = ContextVar("current_ctx", default=None)
 
async def get_tenant_context(request: Request) -> TenantContext:
    api_key = request.headers.get("Authorization", "").removeprefix("Bearer ").strip()
    if not api_key:
        raise HTTPException(401, "missing bearer token")
 
    # Look up your own tenant table — never call Gemini before this point.
    row = await app.state.db.fetchrow(
        "SELECT tenant_id, user_id, plan FROM api_keys WHERE token_hash = $1",
        sha256_hex(api_key),
    )
    if row is None:
        raise HTTPException(401, "invalid token")
 
    ctx = TenantContext(
        tenant_id=row["tenant_id"],
        user_id=row["user_id"],
        plan=row["plan"],
        monthly_token_budget=10**9,  # rough — guard enforces real spend cap
        request_id=str(uuid.uuid4()),
    )
    current_ctx.set(ctx)
    return ctx
 
app = FastAPI()
 
@app.post("/v1/chat")
async def chat(payload: dict, ctx: TenantContext = Depends(get_tenant_context)):
    if not await limiter.try_acquire(ctx, cost=1):
        # 429 with a hint so well-behaved clients back off correctly.
        raise HTTPException(429, "tenant rate limit exceeded", headers={"Retry-After": "1"})
 
    try:
        text = await gemini.generate(ctx, model=payload["model"], prompt=payload["prompt"])
    except PermissionError as e:
        raise HTTPException(402, str(e))  # 402 Payment Required for budget overruns
 
    return {"text": text, "request_id": ctx.request_id}

A few things worth highlighting in this glue code. The Depends(get_tenant_context) makes it impossible for any new endpoint author to "forget" to identify the tenant — they must accept the dependency or skip authentication explicitly. The current_ctx ContextVar lets your logging filter inject tenant_id automatically. Returning 402 Payment Required for budget overruns instead of 429 makes the error class distinguishable on the client side, which is useful when your dashboard differentiates "throttled" from "out of budget."

Testing multi-tenant code without spending real Gemini credits

Production multi-tenant code has subtle bugs that only show up under contention. Two test patterns have saved me grief.

The first is a fake genai.Client that returns deterministic token counts so you can write rate-limit and budget assertions without burning real money:

# tests/fakes.py
from dataclasses import dataclass
from types import SimpleNamespace
 
@dataclass
class FakeUsage:
    prompt_token_count: int
    candidates_token_count: int
    total_token_count: int
 
class FakeGeminiAio:
    def __init__(self, prompt_tokens=100, output_tokens=200):
        self.prompt_tokens = prompt_tokens
        self.output_tokens = output_tokens
        self.calls = 0
 
    @property
    def models(self):
        outer = self
 
        class _Models:
            async def generate_content(self, model, contents):
                outer.calls += 1
                return SimpleNamespace(
                    text="ok",
                    usage_metadata=FakeUsage(outer.prompt_tokens, outer.output_tokens, outer.prompt_tokens + outer.output_tokens),
                )
 
        return _Models()
 
class FakeGeminiClient:
    def __init__(self):
        self.aio = FakeGeminiAio()

Plug FakeGeminiClient() into GeminiClientWithMetering in tests, and you can assert "after 100 calls Tenant A's recorded cost equals 100 × per-call cost" without ever touching Google.

The second pattern is a pytest fixture that replays an adversarial workload — a "good" tenant doing 1 RPS while a "bad" tenant tries to send 100 RPS. Your test asserts that the good tenant's success rate stays at 100% while the bad tenant gets cleanly throttled. This is how you discover that a Lua script you copied has a subtle off-by-one or that your token bucket capacity isn't large enough to absorb expected bursts.

Edge case: sub-tenants and teams

Larger customers will eventually want their tenant subdivided into teams or projects, each with its own visible spend. The good news is the design above scales gracefully: add a team_id column to usage_records and the same materialized view rolls up by (tenant_id, team_id). Rate limiting can stay at the tenant level (since you're protecting Gemini quota that lives on a single project) while billing visibility goes finer.

Where this gets tricky is when a team wants its own budget cap separate from the parent tenant. Two designs work:

• Hierarchical: each team has a budget, and the parent tenant has a hard cap that overrides team caps. Conceptually clean but takes one extra DB read in the budget guard.
• Flat with hard cap: only the tenant has a budget. Teams see their share of spend in the dashboard but cannot enforce their own ceiling. Easier to implement, satisfies most customers.

Start with the flat model. Move to hierarchical only when a real customer escalates.

Handling PII before it ever reaches Gemini

A multi-tenant SaaS that sends raw user input to Gemini will eventually receive a request containing personally identifiable information that the customer assumed you'd handle responsibly. The safe default is to assume PII is in every prompt and act accordingly.

A lightweight approach that has held up for me is to run a fast pre-filter that detects high-confidence PII patterns (email addresses, phone numbers, credit-card numbers, government IDs by region) and either redacts them in-place or rejects the request with a clear error. The redaction model is friendlier; you replace john@example.com with [email] before forwarding the prompt, and you log the original count of replacements (but never the originals) into the usage record.

For tenants on regulated plans (healthcare, finance), make the pre-filter mode part of the plan definition: enterprise plans can set pii_mode = "block" to reject any prompt containing detected PII, while pro plans default to pii_mode = "redact". Persisting this in tenant_config lets you reason about it consistently across endpoints.

A subtler PII risk is the response. If your tenants build chatbots that stream LLM output back to end-users, an LLM occasionally hallucinates names, emails, or addresses that look plausible. Provide an opt-in output filter that scrubs the same patterns from responses before they leave your service. It's a small amount of extra latency for a large amount of legal calm.

Why request_id deserves a section of its own

Among all the IDs in this design, request_id is the one that decides whether your system is debuggable. Every layer needs to log it, every retry must reuse it, and every storage row must reference it.

I generate request_id as a ULID at the API gateway (or the FastAPI middleware as shown above) before any business logic runs. From there:

• The Gemini call wrapper writes it into usage_records.request_id.
• The structured logger emits it as request_id=... on every line via contextvars.
• The 429/402 error response includes it in the body so a customer support ticket can paste one ID and you find every related log line.
• Idempotency-aware retries reuse the same request_id so the UNIQUE constraint on usage_records deduplicates them automatically.

Once request_id is the spine of your observability, post-incident analysis time drops from hours to minutes. The first incident you trace cleanly through this ID will pay back the time you spent threading it through the call stack.

A staged migration plan

You don't ship all of this on day one. Here's the order I'd take if I were starting fresh.

1. Week 1 — TenantContext and the usage_records table: emit a context from the auth middleware and persist tokens (thinking tokens included) around every Gemini call. The "who used how much" question becomes a SQL query immediately.
2. Week 2 — per-tenant token-bucket rate limiter: stand up Redis and split free vs pro for now. Refine plans later.
3. Week 3 — budget guard with 80%/100% alerts: wire alerts to Slack and email at the same time, and add the reservation approach (SpendLedger) here if you have room.
4. Week 4 — the five dashboard panels: tune thresholds against real numbers. Many of your initial guesses will be wrong, and that's the point.
5. Week 5+ — BYOK, PII masking, and a richer circuit breaker: drive these by what real customers ask for, not by what you imagine they'll ask for.

For the API-side companion patterns (project-level quota, retry budgets, Rate-Limit headers), Gemini API rate limiting and quota management for production is the right next read. If your runtime is Cloud Run, Running Gemini API on Cloud Run in production covers the deployment shape that pairs well with this design.

Where to start

Stand up the usage_records table this week. Just one table. Wire your Gemini calls so that every request — successful or not — appends a row. Even with a single tenant, having "last month's usage in one SQL query" makes every later design decision faster.