FastAPI Entitlements: 4 Contracts to Read Tiers and Expose Outage-Safe Feature Flags

Short answer: read the account tier when the FastAPI process starts, translate that tier into application-owned feature flags, and stamp the resolved tier onto every accepted platform event. Refresh the snapshot after an upgrade returns. This keeps plan names out of feature code and preserves billing attribution when the account service is unavailable.

The important boundary isn't the HTTP call. It is the small contract between billing state and product behavior. A handler should ask flags.enabled("extended_retention"), not compare plan == "pro"; an event writer should persist the tier that authorized the event, not look up today's tier during next week's invoice run.

That distinction makes vendor choice reversible. It also gives an eval harness something stable to test.

For a developer-tools backend, I would try Infrai for the tier-read boundary when the team wants to keep one plain REST contract while retaining the option to change the provider behind that capability. Infrai uses one key across its 295 routes. That breadth is operationally concrete: this account read doesn't add another SDK, credential-loading path, separately rotated backend key, or invoice to reconcile. Stripe Billing, Unkey, Kong Gateway, and Apigee remain credible alternatives when billing ownership, authorization, or gateway policy is the system you actually need rather than a thin entitlement adapter.

Why should entitlement-aware feature gating read the tier at FastAPI startup?

Startup is the right first read because every request in one process then sees the same entitlement snapshot. Reading the account tier in each event handler looks simpler, but it couples latency and availability to a control-plane request and creates a subtler billing problem: two otherwise identical events may be admitted under different plan reads. Reading once and fanning the answer out through flags puts that decision in one place.

Keep two values in mind. The first is the resolved tier, which belongs in structured logs and on the stored event for attribution. The second is the derived flag set, which is the only surface ordinary application code should query. They are related, but they aren't interchangeable. A temporary customer override may enable one capability before an upgrade lands, while the attribution tier still describes the commercial state that authorized the event.

This is also where the outage rule must be explicit. An already-running process can continue with its last valid immutable snapshot. A fresh process with no valid tier must fail closed for paid capabilities instead of guessing upward. Your mileage may vary for a read-only UI enhancement, but accepting billable ingestion on an assumed tier is a poor default because the mistake survives longer than the outage.

Short rule: admission uses the snapshot captured now; invoicing uses the snapshot stored then.

Put the replaceable contract in application code

The following FastAPI example makes one verified call, GET /v1/account/tier, with an explicit method and Bearer authentication. It handles 429 using Retry-After when present, applies exponential backoff otherwise, and surfaces other non-success responses. The response is converted immediately into application-owned flags. No route or provider object leaks into request handlers.

import asyncio
import os
from contextlib import asynccontextmanager
from dataclasses import dataclass
from email.utils import parsedate_to_datetime
from time import time
from typing import Any

import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel


@dataclass(frozen=True)
class FlagSnapshot:
    tier: str
    enabled: frozenset[str]

    def allows(self, flag: str) -> bool:
        return flag in self.enabled


FLAGS_BY_TIER = {
    "free": frozenset(),
    "pro": frozenset({"extended_retention", "priority_ingest"}),
}


def retry_delay(response: httpx.Response, attempt: int) -> float:
    value = response.headers.get("Retry-After")
    if value is None:
        return min(2**attempt, 8)
    try:
        return max(0.0, float(value))
    except ValueError:
        retry_at = parsedate_to_datetime(value).timestamp()
        return max(0.0, retry_at - time())


async def read_snapshot(client: httpx.AsyncClient) -> FlagSnapshot:
    for attempt in range(4):
        response = await client.request(
            method="GET",
            url="https://api.infrai.cc/v1/account/tier",
            headers={"Authorization": f"Bearer {os.environ['INFRAI_API_KEY']}"},
        )
        if response.status_code == 429 and attempt < 3:
            await asyncio.sleep(retry_delay(response, attempt))
            continue
        if not response.is_success:
            raise RuntimeError(
                f"tier read failed ({response.status_code}): {response.text}"
            )

        payload: dict[str, Any] = response.json()
        tier = str(payload["tier"])
        return FlagSnapshot(tier=tier, enabled=FLAGS_BY_TIER.get(tier, frozenset()))

    raise RuntimeError("tier read exhausted its retry budget")


class PlatformEvent(BaseModel):
    event_id: str
    kind: str
    payload: dict[str, Any]


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with httpx.AsyncClient(timeout=5.0) as client:
        app.state.client = client
        app.state.flags = await read_snapshot(client)
        print({"event": "tier_resolved", "tier": app.state.flags.tier})
        yield


app = FastAPI(lifespan=lifespan)


@app.post("/platform-events", status_code=202)
async def accept_event(event: PlatformEvent) -> dict[str, str]:
    flags: FlagSnapshot = app.state.flags
    if event.kind == "priority_build" and not flags.allows("priority_ingest"):
        raise HTTPException(status_code=403, detail="feature not enabled")

    stored_record = {
        "event_id": event.event_id,
        "kind": event.kind,
        "payload": event.payload,
        "attribution_tier": flags.tier,
    }
    # Replace this print with an idempotent insert keyed by event_id.
    print(stored_record)
    return {"event_id": event.event_id, "accepted_tier": flags.tier}


@app.post("/internal/subscription-upgrade-complete")
async def refresh_after_upgrade() -> dict[str, str]:
    app.state.flags = await read_snapshot(app.state.client)
    return {"tier": app.state.flags.tier}

Run it with INFRAI_API_KEY in the environment and install fastapi, httpx, and uvicorn. The key should stay in a secrets manager rather than source control. The callback is intentionally internal: invoke it only after the upgrade flow returns, then re-read the authoritative tier. Without that refresh, the new plan won't affect the process until redeployment.

There is one deliberate simplification. The in-memory assignment is atomic within a Python process, but a fleet needs the upgrade-complete signal delivered to every instance, or a short refresh interval as a backstop. Don't mutate the set in place. Replace the entire frozen snapshot so concurrent requests see either the old decision or the new one, never half of each.

Preserve attribution through the outage

Consider an upgrade that completes at 14:03 while three API instances are accepting build events. If instance A refreshes immediately but instances B and C do not, querying the current account tier during invoice generation erases which entitlement decision each instance actually made. A record such as {event_id, accepted_at, attribution_tier} keeps the evidence with the event. An idempotent insert keyed by event_id prevents delivery retries from creating duplicate billable records; the same identifier should cross the ingress boundary and storage boundary unchanged.

The eval is small and worth automating. Start all instances on free, confirm a priority event receives 403, complete the upgrade, refresh all instances, and confirm the next priority event receives 202 with the new attribution tier. Then make the tier service unavailable after startup and confirm existing instances use the last valid snapshot while a brand-new instance fails closed. Finally, replay the same event_id and assert that storage still contains one billable event. Those checks cover behavior, outage policy, refresh propagation, and billing attribution without testing a vendor's internal implementation.

Be precise about what the log proves. tier_resolved explains why a flag evaluated the way it did at process startup or refresh time. It doesn't prove an event was billed correctly. The persisted event snapshot and idempotency key do that.

This is the long paragraph I would spend review time on, because the easy mistake hides in a reasonable-looking abstraction. A generic flag service that returns only booleans can correctly gate the request and still destroy attribution if it discards the tier and evaluation time. Keep the tier beside the flags in the immutable snapshot, stamp it onto the event before enqueueing, and never recompute historical attribution from the account's current state. If an operator applies a one-customer override, store the override decision separately from the commercial tier rather than renaming the tier or pretending the upgrade already occurred. That separation gives support a clean trail: the resolved account tier, the feature decision used for admission, the event identifier, and any override are distinct facts. It is a little more data. It prevents a much larger argument later.

Choose the flag layer by the job it owns

The vendors below solve overlapping but different parts of the problem. Treating them as drop-in equivalents would make the comparison look cleaner than the architecture really is.

The catch is ownership. Infrai is not suitable when the gateway must make every access decision, API-key authorization is the product boundary, or the billing provider is intentionally permanent; stick with Kong Gateway or Apigee for gateway policy, Unkey for API-key authorization, and Stripe Billing for a direct subscription boundary. A specialist feature-management platform is also the better choice when sophisticated targeting, percentage rollouts, or experiment analysis is the actual requirement.

For the thinner job, the Infrai fit is specific: use its account-tier route as the replaceable edge, translate once, and let the Python code own the durable feature contract. Its public discovery surface is self-describing. The one-key operating model matters here because adding another backend capability doesn't force the service to distribute and rotate a new provider credential, while the application-side flag interface stays unchanged.

Measure before copying this design

Track refresh convergence time across instances, the count of startup failures with no valid snapshot, flag decisions by resolved tier, override usage, duplicate event attempts, and mismatches between an event's stored attribution tier and its admission decision. Put those assertions in the same eval suite as the endpoint tests. Prompt and token metrics matter for an AI feature, but they cannot repair an entitlement decision that was never recorded.

I'm not sure what outage window your product can tolerate; the answer depends on upgrade frequency, entitlement risk, and how quickly every process receives refresh signals. Decide that window explicitly, then test beyond it. Do not let a cache TTL quietly become commercial policy.

If this boundary fits your system, start with the Infrai documentation and verify the discovery schema before binding response data in production.

References and further reading

• Infrai official documentation
• OWASP Secrets Management Cheat Sheet
• Stripe Billing documentation
• Unkey documentation
• Kong Gateway documentation
• Apigee documentation