Operations

What it's for: switches flipped at runtime, without a deploy: maintenance mode, features rolled out to some of the users, and a breaker for a service that's down.

An app operated at runtime, without a deploy, through a small admin API that takes a bearer token. It closes and reopens the app with FastAPIMaintenance, whose middleware answers 503 with Retry-After to everything but the health check and the admin routes. It defines FastAPIFeatureFlags: a new checkout rolled out to a percentage of the users, and a beta page that only the users on an allow list can see (a 404 for everybody else). And a FastAPIBreaker turns the failures of a flaky shipping service into a fast 503 while it's down.

The maintenance switch and the flags live in the backend, so a change made through one process reaches all of them. Each process keeps a copy for up to 1 s (maintenance) or 5 s (flags), so that requests don't each ask the backend. The breaker is each process's own.

Run it

From a clone of the repository:

make deps                 # .venv, with every extra
source .venv/bin/activate
uvicorn examples.operations.main:app

The admin routes take Authorization: Bearer $ADMIN_TOKEN. Set ADMIN_TOKEN and CACHE_SIGNING_KEY (secrets.token_urlsafe(32)), the same in every process. Without ADMIN_TOKEN, the app makes one up at each start and logs it:

ADMIN_TOKEN isn't set: the admin token of this run is YbLt_8-WKnm4zWknL3mgHNar4E3MmBsCNtk6KPJLBB4

It keeps the switch and the flags in memory, so don't start it with --workers: each worker would have a switch of its own. In production, a Redis or Valkey backend lets every process share them.

Try it

$ TOKEN=YbLt_8-WKnm4zWknL3mgHNar4E3MmBsCNtk6KPJLBB4

Without the token, the admin routes answer 401:

$ curl -i -X PUT http://127.0.0.1:8000/admin/maintenance -H "Content-Type: application/json" -d '{}'
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer

{"detail":"Unauthorized"}

Close the app for maintenance. Every request gets a 503 with the message and Retry-After, but the health check is still served:

$ curl -X PUT http://127.0.0.1:8000/admin/maintenance -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" -d '{"message": "Back at 10:30", "retry_after": 900}'
{"enabled":true,"message":"Back at 10:30","retry_after":900,"since":1793011967.4372241}

$ curl -i http://127.0.0.1:8000/checkout
HTTP/1.1 503 Service Unavailable
cache-control: no-store
retry-after: 900

{"detail": "Back at 10:30"}

$ curl http://127.0.0.1:8000/health
{"status":"ok"}

$ curl -i -X DELETE http://127.0.0.1:8000/admin/maintenance -H "Authorization: Bearer $TOKEN"
HTTP/1.1 204 No Content

$ curl http://127.0.0.1:8000/checkout
{"checkout":"classic"}

Roll the new checkout out to 30 % of the users. Here the bearer token stands for the user, so that you can try it with curl. Each user keeps their answer from one request to the next, and anonymous visitors aren't part of a partial rollout:

$ curl -X PUT http://127.0.0.1:8000/admin/flags/new-checkout -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" -d '{"enabled": true, "percentage": 30}'
{"name":"new-checkout","enabled":true,"percentage":30.0,"allow":[],"deny":[]}

$ for user in ana bea carlos david elena; do \
    echo "$user: $(curl -s http://127.0.0.1:8000/checkout -H "Authorization: Bearer $user")"; \
  done
ana: {"checkout":"new"}
bea: {"checkout":"classic"}
carlos: {"checkout":"classic"}
david: {"checkout":"new"}
elena: {"checkout":"classic"}

$ curl http://127.0.0.1:8000/checkout
{"checkout":"classic"}

The beta report is a 404, like a page that doesn't exist, until the user is on the allow list. Without a percentage, a flag with an allow list is for nobody else. The list comes back as digests: the backend never gets the users' ids.

$ curl http://127.0.0.1:8000/reports/beta -H "Authorization: Bearer ana"
{"detail":"Not Found"}

$ curl -X PUT http://127.0.0.1:8000/admin/flags/beta-reports -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" -d '{"enabled": true, "allow": ["ana"]}'
{"name":"beta-reports","enabled":true,"percentage":null,"allow":["44d27c2ae32d479abf13f231ea9ee8295988d4a14fdf8be30c16aa81e894946b"],"deny":[]}

$ curl http://127.0.0.1:8000/reports/beta -H "Authorization: Bearer ana"
{"report":"Sales by hour"}

$ curl -i http://127.0.0.1:8000/reports/beta -H "Authorization: Bearer bea"
HTTP/1.1 404 Not Found

{"detail":"Not Found"}

Make the shipping service fail. The first three requests wait for it and get a 502; then the breaker opens, and requests get a 503 at once, without calling the service:

$ curl http://127.0.0.1:8000/orders/7/shipping
{"order_id":7,"carrier":"Alhambra Express","days":2}

$ curl -X PUT "http://127.0.0.1:8000/admin/simulate/shipping?down=true" -H "Authorization: Bearer $TOKEN"

$ curl -i http://127.0.0.1:8000/orders/7/shipping    # three times
HTTP/1.1 502 Bad Gateway

{"detail":"The shipping service failed"}

$ curl -i http://127.0.0.1:8000/orders/7/shipping
HTTP/1.1 503 Service Unavailable
retry-after: 30

{"detail":"Service Unavailable"}

The status page reads the shared state from the backend, and the breaker from this process:

$ curl http://127.0.0.1:8000/admin/status -H "Authorization: Bearer $TOKEN"
{"maintenance":{"enabled":false,"message":null,"retry_after":null,"since":null},"flags":{"new-checkout":{...},"beta-reports":{...}},"process":{"pid":831700,"shipping_breaker":"open"}}

The breaker stays open for 30 s, even when the service is back. Then a single request tries it, and its success closes the breaker:

$ curl -X PUT "http://127.0.0.1:8000/admin/simulate/shipping?down=false" -H "Authorization: Bearer $TOKEN"

$ curl -i http://127.0.0.1:8000/orders/7/shipping
HTTP/1.1 503 Service Unavailable
retry-after: 30

{"detail":"Service Unavailable"}

$ sleep 30; curl http://127.0.0.1:8000/orders/7/shipping
{"order_id":7,"carrier":"Alhambra Express","days":2}

Or, once you know the service is back, close the breaker at once. Only in the process that gets the request: each process has its own.

$ curl -i -X POST http://127.0.0.1:8000/admin/breakers/shipping/close -H "Authorization: Bearer $TOKEN"
HTTP/1.1 204 No Content

$ curl http://127.0.0.1:8000/orders/7/shipping
{"order_id":7,"carrier":"Alhambra Express","days":2}

What to look at

• The admin token is compared in constant time, as bytes: secrets.compare_digest takes the same time however much of a guess is right. It compares bytes because on strings it raises when a client sends non-ASCII characters: a 500 instead of a 401.
• Shared state, local copies: the switch and the flags are in the backend, signed with CACHE_SIGNING_KEY, so a state planted there by someone else is ignored. Each process reads them again at most every refresh seconds, so a change takes up to 1 s (maintenance) or 5 s (flags) to reach the other processes; the process that made it, and /admin/status, see it at once. The breaker is created in the lifespan, one per process: it opens after that process's failures, with no round trip on every call.
• /health and /admin are always served: the orchestrator doesn't restart processes that are only in maintenance, and the API that closed the app can reopen it.
• Flag names are an enum: no flag the app never reads can be created. A flag with an allow list and no percentage is only for the list, so a beta never goes to everyone by mistake.
• Subjects come from authentication: a route hidden by flags.require() takes its subject from a function of the request. Here the bearer token is the user id; in a real app take it from a session or a token you verify, or anyone could put themselves on an allow list.
• Only what says the service is down opens the breaker: failures=(ConnectionError,), so a bug of ours doesn't, and a call that hangs is cancelled after 5 s and counts too. A failure is a 502; a refusal while the breaker is open, a 503 with Retry-After.

The code

examples/operations/main.py

"""Operating an app without a deploy: maintenance mode, feature flags and a circuit breaker.

    uvicorn examples.operations.main:app

- `PUT /admin/maintenance` closes the app: a 503 with `Retry-After` for every request but
  `/health` (the load balancer keeps the processes) and `/admin` (you can reopen it).
- `PUT /admin/flags/{name}`: `new-checkout` goes to a percentage of the users, always the same
  ones; `/reports/beta` is a 404 but for the users on the allow list of `beta-reports`.
- `/orders/{order_id}/shipping` calls a flaky shipping service (simulated). After 3 failures in a
  row, each a 502 that waited for it, the breaker opens: a 503 at once for 30 s, without calling
  it. Then one request tries again; `POST /admin/breakers/shipping/close` closes it at once.

The admin routes take `Authorization: Bearer $ADMIN_TOKEN`.

The maintenance switch and the flags live in the backend, shared by every process. Each process
keeps a copy, read again at most every 1 s (maintenance) or 5 s (flags): requests don't each ask
the backend, and a change reaches every process within that. Breakers are each process's own:
one opens after its process's failures, with no round trip on every call.
"""

import logging
import os
import secrets
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from enum import StrEnum
from typing import Annotated, Any

import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.requests import HTTPConnection

from fastapi_alcazaba import (
    BackendUnavailable,
    FastAPIBreaker,
    FastAPICache,
    FastAPIFeatureFlags,
    FastAPIMaintenance,
)
from fastapi_alcazaba.backends.inmemory import InMemoryBackend
from fastapi_alcazaba.operations import MAX_MESSAGE_LENGTH, FeatureFlag, MaintenanceMiddleware
from fastapi_alcazaba.operations.maintenance import MaintenanceStatus

logger = logging.getLogger(__name__)


def throwaway_secret() -> SecretStr:
    """A random secret, for trying the example: it changes at every start and differs between
    processes."""
    return SecretStr(secrets.token_urlsafe(32))


class Settings(BaseSettings):
    """The configuration, from environment variables or a .env file. Set both in production, the
    same in every process (secrets.token_urlsafe(32))."""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")

    admin_token: SecretStr = Field(default_factory=throwaway_secret)
    # Seals the switch and the flags in the backend: a state planted there by someone else is
    # ignored, and so is what a process with another key writes.
    cache_signing_key: SecretStr = Field(default_factory=throwaway_secret)


settings = Settings()


class Flag(StrEnum):
    """The flags this app reads: the admin routes define only these."""

    NEW_CHECKOUT = "new-checkout"
    BETA_REPORTS = "beta-reports"


class Quote(BaseModel):
    order_id: int
    carrier: str
    days: int


class ShippingService:
    """Stands in for a downstream service and its SDK: fails while `down`, and counts the calls
    that reach it."""

    def __init__(self) -> None:
        self.down = False
        self.calls = 0

    async def quote(self, order_id: int) -> Quote:
        self.calls += 1
        if self.down:
            raise ConnectionError("the shipping service doesn't answer")
        return Quote(order_id=order_id, carrier="Alhambra Express", days=2)


@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
    # Redis or Valkey in production, so that every process shares the switch and the flags:
    # InMemoryBackend keeps them in this process.
    FastAPICache.init(InMemoryBackend(), prefix="ops", signing_key=settings.cache_signing_key)
    application.state.shipping = ShippingService()
    # Only what says the service is down counts, not a bug of ours; a call that hangs is cancelled
    # after 5 s and counts too.
    application.state.shipping_breaker = FastAPIBreaker(
        "shipping", failure_threshold=3, reset_timeout=30, timeout=5, failures=(ConnectionError,)
    )
    if "admin_token" not in settings.model_fields_set:
        token = settings.admin_token.get_secret_value()
        logger.warning("ADMIN_TOKEN isn't set: the admin token of this run is %s", token)
    yield


# How old each process's copy of the shared state may be, in seconds.
maintenance = FastAPIMaintenance(refresh=1)
flags = FastAPIFeatureFlags(refresh=5)

app = FastAPI(lifespan=lifespan)
app.add_middleware(
    MaintenanceMiddleware, maintenance=maintenance, allow_paths=("/health", "/admin")
)


@app.exception_handler(BackendUnavailable)
async def backend_unavailable(_request: Request, _exc: BackendUnavailable) -> JSONResponse:
    """An admin change the backend didn't take: nothing changed, and the response says so."""
    return JSONResponse({"detail": "Not changed: the backend is unavailable"}, status_code=503)


def current_user(connection: HTTPConnection) -> str | None:
    """The user's id. Stands in for your authentication: here the bearer token is the id itself,
    so that you can try it with curl. Take it from a session or a token you verify, or anyone
    could put themselves on an allow list."""
    scheme, _, token = connection.headers.get("authorization", "").partition(" ")
    return token if scheme.lower() == "bearer" and token else None


@app.get("/health")
async def health() -> dict[str, str]:
    """Served during maintenance too: the processes are fine, nothing should restart them."""
    return {"status": "ok"}


@app.get("/checkout")
async def checkout(user: Annotated[str | None, Depends(current_user)]) -> dict[str, str]:
    """The new checkout for the users in the rollout; without a user, a partial rollout is off."""
    new = await flags.enabled(Flag.NEW_CHECKOUT, subject=user)
    return {"checkout": "new" if new else "classic"}


@app.get(
    "/reports/beta",
    dependencies=[Depends(flags.require(Flag.BETA_REPORTS, subject=current_user))],
)
async def beta_report() -> dict[str, str]:
    """A 404 for the users the flag is off for, as if the route didn't exist."""
    return {"report": "Sales by hour"}


@app.get("/orders/{order_id}/shipping")
async def shipping_quote(order_id: int, request: Request) -> Quote:
    """A 502 when the service fails; while the breaker is open, a 503 with Retry-After."""
    breaker: FastAPIBreaker = request.app.state.shipping_breaker
    shipping: ShippingService = request.app.state.shipping
    try:
        async with breaker.attempt():
            return await shipping.quote(order_id)
    except (ConnectionError, TimeoutError) as exc:
        raise HTTPException(status.HTTP_502_BAD_GATEWAY, "The shipping service failed") from exc


def require_admin(
    credentials: Annotated[
        HTTPAuthorizationCredentials | None, Depends(HTTPBearer(auto_error=False))
    ],
) -> None:
    """Compared in constant time, and as bytes: compare_digest refuses strings with non-ASCII
    characters, which a client can send."""
    token = credentials.credentials if credentials is not None else ""
    if not secrets.compare_digest(token.encode(), settings.admin_token.get_secret_value().encode()):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, headers={"WWW-Authenticate": "Bearer"})


admin = APIRouter(prefix="/admin", dependencies=[Depends(require_admin)])


class MaintenanceUpdate(BaseModel):
    message: str | None = Field(default=None, max_length=MAX_MESSAGE_LENGTH)
    retry_after: int | None = Field(default=None, ge=1)


class FlagUpdate(BaseModel):
    enabled: bool
    # None: everybody, or only the allow list when there's one.
    percentage: float | None = Field(default=None, ge=0, le=100)
    allow: list[str] = Field(default=[], max_length=1000)
    deny: list[str] = Field(default=[], max_length=1000)


@admin.put("/maintenance")
async def start_maintenance(update: MaintenanceUpdate) -> MaintenanceStatus:
    """Close the app, in every process within a second."""
    return await maintenance.enable(update.message, update.retry_after)


@admin.delete("/maintenance", status_code=status.HTTP_204_NO_CONTENT)
async def end_maintenance() -> None:
    await maintenance.disable()


@admin.put("/flags/{name}")
async def set_flag(name: Flag, update: FlagUpdate) -> FeatureFlag:
    """Define a flag, for every process within 5 seconds. The lists come back as digests: the
    backend never gets the users' ids."""
    return await flags.set(name, **update.model_dump())


@admin.put("/simulate/shipping", status_code=status.HTTP_204_NO_CONTENT)
async def simulate_shipping(down: bool, request: Request) -> None:
    """Only in this example: makes the shipping service fail, to watch the breaker open."""
    request.app.state.shipping.down = down


@admin.post("/breakers/shipping/close", status_code=status.HTTP_204_NO_CONTENT)
async def close_shipping_breaker(request: Request) -> None:
    """The shipping service is back: call it again now, instead of waiting for the trial. Only
    in the process that gets this request: each process has its own breaker."""
    request.app.state.shipping_breaker.reset()


@admin.get("/status")
async def status_report(request: Request) -> dict[str, Any]:
    """The shared state as the backend has it now (this process's copies are updated with it),
    and the breaker as this process sees it."""
    breaker: FastAPIBreaker = request.app.state.shipping_breaker
    return {
        "maintenance": await maintenance.status(),
        "flags": {flag: await flags.get(flag) for flag in Flag},
        "process": {"pid": os.getpid(), "shipping_breaker": breaker.state},
    }


app.include_router(admin)


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)