FastAPI Adapter¶
The FastAPI adapter is the currently supported ASGI adapter. It provides middleware, decorators, a drop-in router, and OpenAPI integration — all built on top of the framework-agnostic shield.core.
Other ASGI frameworks
api-shield's core and ShieldMiddleware are ASGI-native and framework-agnostic. FastAPI-specific features (ShieldRouter, OpenAPI integration, Depends() support) live in shield.fastapi. Adapters for Litestar and plain Starlette are on the roadmap. Open an issue if you need another framework prioritised.
Installation¶
uv add "api-shield[fastapi]" # adapter only
uv add "api-shield[fastapi,rate-limit]" # with rate limiting
uv add "api-shield[all]" # everything including CLI + admin
Quick setup¶
main.py
from fastapi import FastAPI
from shield.core.config import make_engine
from shield.fastapi import (
ShieldMiddleware,
ShieldAdmin,
apply_shield_to_openapi,
setup_shield_docs,
maintenance,
env_only,
disabled,
force_active,
deprecated,
)
engine = make_engine() # reads SHIELD_BACKEND, SHIELD_ENV from env / .shield
app = FastAPI()
app.add_middleware(ShieldMiddleware, engine=engine)
@app.get("/payments")
@maintenance(reason="DB migration")
async def get_payments():
return {"payments": []}
@app.get("/health")
@force_active
async def health():
return {"status": "ok"}
apply_shield_to_openapi(app, engine)
setup_shield_docs(app, engine)
app.mount("/shield", ShieldAdmin(engine=engine, auth=("admin", "secret")))
Components¶
ShieldMiddleware¶
ASGI middleware that enforces route state on every request.
See Reference: Middleware for full details.
Decorators¶
All decorators work with any router type (plain APIRouter, ShieldRouter, or routes added directly to app).
| Decorator | Import | Behaviour |
|---|---|---|
@maintenance(reason, start, end) |
shield.fastapi |
503 temporarily |
@disabled(reason) |
shield.fastapi |
503 permanently |
@env_only(*envs) |
shield.fastapi |
404 in other envs |
@deprecated(sunset, use_instead) |
shield.fastapi |
200 + headers |
@force_active |
shield.fastapi |
Always 200 |
@rate_limit("100/minute") |
shield.fastapi.decorators |
429 when exceeded |
See Reference: Decorators for full details.
ShieldRouter¶
A drop-in replacement for APIRouter that automatically registers route metadata with the engine at startup.
from shield.fastapi.router import ShieldRouter
router = ShieldRouter(engine=engine)
@router.get("/payments")
@maintenance(reason="DB migration")
async def get_payments():
return {"payments": []}
app.include_router(router)
Note
ShieldRouter is optional. ShieldMiddleware also registers routes by scanning app.routes at startup (lazy, on first request). Use ShieldRouter for explicit control over registration order.
ShieldAdmin¶
Mounts the admin dashboard UI and the REST API (used by the shield CLI) under a single path.
from shield.admin import ShieldAdmin
app.mount("/shield", ShieldAdmin(engine=engine, auth=("admin", "secret")))
See Tutorial: Admin Dashboard for full details.
Rate limiting¶
Requires api-shield[rate-limit] on the server.
from shield.fastapi.decorators import rate_limit
@router.get("/public/posts")
@rate_limit("10/minute") # 10 req/min per IP
async def list_posts():
return {"posts": []}
@router.get("/users/me")
@rate_limit("100/minute", key="user") # per authenticated user
async def get_current_user():
...
@router.get("/reports")
@rate_limit( # tiered limits
{"free": "10/minute", "pro": "100/minute", "enterprise": "unlimited"},
key="user",
)
async def get_reports():
...
Custom response on rate limit violations:
from starlette.requests import Request
from starlette.responses import JSONResponse
from shield.core.exceptions import RateLimitExceededException
def my_429(request: Request, exc: RateLimitExceededException) -> JSONResponse:
return JSONResponse(
{"ok": False, "retry_after": exc.retry_after_seconds},
status_code=429,
headers={"Retry-After": str(exc.retry_after_seconds)},
)
@router.get("/posts")
@rate_limit("10/minute", response=my_429)
async def list_posts(): ...
Global default (applies to all rate-limited routes without a per-route factory):
Mutate policies at runtime without redeploying (shield rl and shield rate-limits are aliases):
See Tutorial: Rate Limiting and Reference: Rate Limiting for full details.
Dependency injection¶
Shield decorators work as FastAPI Depends() dependencies for per-handler enforcement without middleware.
two patterns
from fastapi import Depends
from shield.fastapi.decorators import disabled, maintenance
# Pattern A — decorator (relies on ShieldMiddleware to enforce)
@router.get("/payments")
@maintenance(reason="DB migration")
async def get_payments():
return {"payments": []}
# Pattern B — Depends() only (per-handler, no middleware required)
@router.get("/admin/report", dependencies=[Depends(disabled(reason="Use /v2/report"))])
async def admin_report():
return {}
@router.get("/orders", dependencies=[Depends(maintenance(reason="Order upgrade"))])
async def get_orders():
return {"orders": []}
@rate_limit also works as a Depends():
from shield.fastapi.decorators import rate_limit
@router.get("/export", dependencies=[Depends(rate_limit("5/hour", key="user"))])
async def export():
...
Both the decorator path and the Depends() path share the same counter — they are equivalent in enforcement.
| Pattern | Best for |
|---|---|
| Decorator | Apps that always run ShieldMiddleware |
Depends() |
Serverless / edge runtimes without middleware, or when middleware is not used |
Using with FastAPI lifespan¶
main.py
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
async with engine: # calls backend.startup() then backend.shutdown()
yield
app = FastAPI(lifespan=lifespan)
app.add_middleware(ShieldMiddleware, engine=engine)
Testing¶
tests/test_payments.py
import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from shield.core.backends.memory import MemoryBackend
from shield.core.engine import ShieldEngine
from shield.fastapi.decorators import maintenance, force_active
from shield.fastapi.middleware import ShieldMiddleware
from shield.fastapi.router import ShieldRouter
async def test_maintenance_returns_503():
engine = ShieldEngine(backend=MemoryBackend())
app = FastAPI()
app.add_middleware(ShieldMiddleware, engine=engine)
router = ShieldRouter(engine=engine)
@router.get("/payments")
@maintenance(reason="DB migration")
async def get_payments():
return {"ok": True}
app.include_router(router)
await app.router.startup() # trigger shield route registration
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as client:
resp = await client.get("/payments")
assert resp.status_code == 503
assert resp.json()["error"]["code"] == "MAINTENANCE_MODE"
async def test_runtime_enable_via_engine():
engine = ShieldEngine(backend=MemoryBackend())
await engine.set_maintenance("GET:/orders", reason="Upgrade")
await engine.enable("GET:/orders")
state = await engine.get_state("GET:/orders")
assert state.status.value == "active"
In pyproject.toml:
Runnable examples¶
Each example below is a complete, self-contained FastAPI app. Click to expand the full source, then copy and run it locally.
Basic usage¶
All core decorators + ShieldAdmin
Demonstrates every decorator (@maintenance, @disabled, @env_only, @force_active, @deprecated) together with the ShieldAdmin unified interface (dashboard + CLI REST API).
Expected behavior:
| Endpoint | Response | Why |
|---|---|---|
GET /health |
200 always | @force_active |
GET /payments |
503 MAINTENANCE_MODE |
@maintenance |
GET /debug |
200 in dev, 404 in production | @env_only("dev") |
GET /old-endpoint |
503 ROUTE_DISABLED |
@disabled |
GET /v1/users |
200 + Deprecation headers |
@deprecated |
Run:
uv run uvicorn examples.fastapi.basic:app --reload
# Swagger UI: http://localhost:8000/docs
# Admin dashboard: http://localhost:8000/shield/ (admin / secret)
# Audit log: http://localhost:8000/shield/audit
CLI quick-start:
shield login admin # password: secret
shield status
shield disable GET:/payments --reason "hotfix"
shield enable GET:/payments
Full source:
examples/fastapi/basic.py
"""FastAPI — Basic Usage Example.
Demonstrates the core api-shield decorators together with the ShieldAdmin
unified admin interface (dashboard UI + REST API for the CLI).
Run:
uv run uvicorn examples.fastapi.basic:app --reload
Then visit:
http://localhost:8000/docs — filtered Swagger UI
http://localhost:8000/shield/ — admin dashboard (login: admin / secret)
http://localhost:8000/shield/audit — audit log
CLI quick-start (auto-discovers the server URL):
shield login admin # password: secret
shield status
shield disable /payments --reason "hotfix"
shield enable /payments
Expected behaviour (dev env — set APP_ENV=production to see /debug return 404):
GET /health → 200 always (@force_active)
GET /payments → 503 MAINTENANCE_MODE (@maintenance)
GET /debug → 200 (@env_only("dev"), allowed in dev)
GET /old-endpoint → 503 ROUTE_DISABLED (@disabled)
GET /v1/users → 200 + deprecation headers (@deprecated)
Switch to dev to unlock /debug:
APP_ENV=dev uv run uvicorn examples.fastapi.basic:app --reload
"""
import os
from fastapi import FastAPI
from shield.admin import ShieldAdmin
from shield.core.config import make_engine
from shield.fastapi import (
ShieldMiddleware,
ShieldRouter,
apply_shield_to_openapi,
deprecated,
disabled,
env_only,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# Routes with shield decorators
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health():
"""Always 200 — bypasses every shield check."""
return {"status": "ok", "env": CURRENT_ENV}
@router.get("/payments")
@maintenance(reason="Scheduled database migration — back at 04:00 UTC")
async def get_payments():
"""Returns 503 MAINTENANCE_MODE."""
return {"payments": []}
@router.get("/debug")
@env_only("dev")
async def debug():
"""Returns silent 404 in production. Set APP_ENV=dev to unlock."""
return {"debug": True, "env": CURRENT_ENV}
@router.get("/old-endpoint")
@disabled(reason="Use /v2/endpoint instead")
async def old_endpoint():
"""Returns 503 ROUTE_DISABLED."""
return {}
@router.get("/v1/users")
@deprecated(sunset="Sat, 01 Jan 2027 00:00:00 GMT", use_instead="/v2/users")
async def v1_users():
"""Returns 200 with Deprecation, Sunset, and Link response headers."""
return {"users": [{"id": 1, "name": "Alice"}]}
@router.get("/v2/users")
async def v2_users():
"""Active successor to /v1/users."""
return {"users": [{"id": 1, "name": "Alice"}]}
# ---------------------------------------------------------------------------
# App assembly
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Basic Example",
description=(
"Core decorators: `@maintenance`, `@disabled`, `@env_only`, "
"`@force_active`, `@deprecated`.\n\n"
f"Current environment: **{CURRENT_ENV}**"
),
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
apply_shield_to_openapi(app, engine)
# Mount the unified admin interface:
# - Dashboard UI → http://localhost:8000/shield/
# - REST API → http://localhost:8000/shield/api/... (used by the CLI)
#
# auth= accepts:
# ("user", "pass") — single user
# [("alice","a1"),("bob","b2")] — multiple users
# MyAuthBackend() — custom ShieldAuthBackend subclass
#
# secret_key= should be a stable value in production so tokens survive
# process restarts. Omit it (or set to None) in development — a random key
# is generated on each startup, invalidating all sessions on restart.
app.mount(
"/shield",
ShieldAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/shield",
# secret_key="change-me-in-production",
# token_expiry=86400, # seconds — default 24 h
),
)
Dependency injection¶
Shield decorators as FastAPI Depends()
Shows how to use shield decorators as Depends() instead of (or alongside) middleware. Once configure_shield(app, engine) is called — or ShieldMiddleware is added, which calls it automatically — all decorator dependencies find the engine via request.app.state without needing an explicit engine= argument per route.
Expected behavior:
| Endpoint | Response |
|---|---|
GET /payments |
503 (maintenance) — toggle off with shield enable GET:/payments |
GET /old-endpoint |
503 (disabled) |
GET /debug |
404 in production, 200 in dev/staging |
GET /v1/users |
200 + Deprecation / Sunset / Link headers |
GET /health |
200 always |
Run:
uv run uvicorn examples.fastapi.dependency_injection:app --reload
# Admin dashboard: http://localhost:8000/shield/ (admin / secret)
Try it:
curl -i http://localhost:8000/payments # → 503
shield enable GET:/payments # toggle off without redeploy
curl -i http://localhost:8000/payments # → 200
Full source:
examples/fastapi/dependency_injection.py
"""FastAPI — Dependency Injection Example.
Shows how to use shield decorators as FastAPI ``Depends()`` dependencies
instead of (or alongside) the middleware model — but not both on the
same route. Pick one: decorator (with ShieldMiddleware) or Depends()
(without middleware).
Call ``configure_shield(app, engine)`` once and all decorator deps find the
engine automatically via ``request.app.state.shield_engine`` — no ``engine=``
argument per route. ``ShieldMiddleware`` calls ``configure_shield``
automatically at ASGI startup, so if you use middleware you don't need to
call it manually.
Use either the decorator (with ShieldMiddleware) or ``Depends()`` (without
middleware) — not both on the same route.
Decorator support as ``Depends()``:
✅ maintenance — raises 503 when route is in maintenance
✅ disabled — raises 503 when route is disabled
✅ env_only — raises 404 when accessed from the wrong environment
✅ deprecated — injects Deprecation / Sunset / Link headers on the response
❌ force_active — decorator-only; shield checks run in the middleware, which
completes before any dependency is resolved. A dependency
has no mechanism to retroactively bypass that check.
Run:
uv run uvicorn examples.fastapi.dependency_injection:app --reload
Admin dashboard:
http://localhost:8000/shield/ — login: admin / secret
CLI quick-start:
shield login admin # password: secret
shield status # see all route states
shield enable /payments # toggle off maintenance without redeploy
shield disable /payments --reason "emergency patch"
Try these requests:
curl -i http://localhost:8000/payments # → 503 MAINTENANCE_MODE
shield enable /payments # toggle off without redeploy
curl -i http://localhost:8000/payments # → 200
curl -i http://localhost:8000/old-endpoint # → 503 ROUTE_DISABLED
curl -i http://localhost:8000/debug # → 404 in production env; set APP_ENV=production
curl -i http://localhost:8000/v1/users # → 200 + Deprecation headers
curl -i http://localhost:8000/health # → 200 always
"""
import os
from fastapi import Depends, FastAPI
from shield.admin import ShieldAdmin
from shield.core.config import make_engine
from shield.fastapi import (
ShieldMiddleware,
ShieldRouter,
apply_shield_to_openapi,
deprecated,
disabled,
env_only,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# App assembly — configure_shield is called automatically by ShieldMiddleware
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Dependency Injection Example",
description=(
"``configure_shield(app, engine)`` called once — no ``engine=`` per route.\n\n"
f"Current environment: **{CURRENT_ENV}**"
),
)
# ShieldMiddleware auto-calls configure_shield(app, engine) at ASGI startup.
# Without middleware: from shield.fastapi import configure_shield
# configure_shield(app, engine)
app.add_middleware(ShieldMiddleware, engine=engine)
# ---------------------------------------------------------------------------
# Routes — engine resolved from app.state; no engine= needed per route
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health():
"""Always 200."""
return {"status": "ok", "env": CURRENT_ENV}
@router.get("/users")
async def list_users():
return {"users": [{"id": 1, "name": "Alice"}]}
# Depends() — enforces at the handler level without requiring ShieldMiddleware.
@router.get(
"/payments",
dependencies=[Depends(maintenance(reason="Scheduled DB migration"))],
)
async def get_payments():
"""503 on startup; toggle off with: shield enable /payments"""
return {"payments": []}
@router.get(
"/old-endpoint",
dependencies=[Depends(disabled(reason="Use /v2/endpoint instead"))],
)
async def old_endpoint():
"""503 on startup; re-enable with: shield enable /old-endpoint"""
return {}
@router.get(
"/debug",
dependencies=[Depends(env_only("dev", "staging"))],
)
async def debug():
"""404 in production; 200 in dev/staging."""
return {"env": CURRENT_ENV}
# @deprecated as a Depends() — injects Deprecation, Sunset, and Link headers
# directly on the response at the handler level.
@router.get(
"/v1/users",
dependencies=[
Depends(
deprecated(
sunset="Sat, 01 Jan 2027 00:00:00 GMT",
use_instead="/v2/users",
)
)
],
)
async def v1_users():
"""200 always, but carries Deprecation + Sunset + Link response headers."""
return {"users": [{"id": 1, "name": "Alice"}]}
@router.get("/v2/users")
async def v2_users():
"""Active successor to /v1/users."""
return {"users": [{"id": 1, "name": "Alice"}]}
# @force_active cannot be used as a Depends() — see module docstring for why.
# It is applied as a decorator only.
app.include_router(router)
apply_shield_to_openapi(app, engine)
# ---------------------------------------------------------------------------
# Admin interface — dashboard UI + REST API (used by the CLI)
# ---------------------------------------------------------------------------
app.mount(
"/shield",
ShieldAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/shield",
# secret_key="change-me-in-production",
),
)
Scheduled maintenance¶
Auto-activating and auto-deactivating windows
Demonstrates how to schedule a maintenance window that activates and deactivates automatically at the specified times — no manual intervention required.
Endpoints:
| Endpoint | Purpose |
|---|---|
GET /orders |
Normal route — enters maintenance during the window |
GET /admin/schedule |
Schedules a 10-second window starting 5 seconds from now |
GET /admin/status |
Current shield state for all routes |
GET /health |
Always 200 |
Run:
Quick demo:
# 1. Route is active
curl http://localhost:8000/orders # → 200
# 2. Schedule the window (activates in 5 s, ends in 15 s)
curl http://localhost:8000/admin/schedule
# 3. Wait 5 seconds, then:
curl http://localhost:8000/orders # → 503 MAINTENANCE_MODE
# 4. Wait 10 more seconds, then:
curl http://localhost:8000/orders # → 200 again
Full source:
examples/fastapi/scheduled_maintenance.py
"""FastAPI — Scheduled Maintenance Window Example.
Demonstrates how to schedule a future maintenance window that auto-activates
and auto-deactivates at the specified times.
Run:
uv run uvicorn examples.fastapi.scheduled_maintenance:app --reload
Endpoints:
GET /orders — active normally; enters maintenance during the window
GET /admin/schedule — schedules a maintenance window 5 seconds from now
GET /admin/status — shows current route states
GET /health — always 200 (@force_active)
Quick demo:
1. Open http://localhost:8000/orders → 200
2. Hit http://localhost:8000/admin/schedule → schedules window
3. Wait ~5 seconds
4. Hit http://localhost:8000/orders → 503 MAINTENANCE_MODE
5. Wait another 10 seconds (window ends)
6. Hit http://localhost:8000/orders → 200 again
"""
from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from fastapi import FastAPI
from shield.core.backends.memory import MemoryBackend
from shield.core.engine import ShieldEngine
from shield.core.models import MaintenanceWindow
from shield.fastapi import ShieldMiddleware, ShieldRouter, force_active
engine = ShieldEngine(backend=MemoryBackend())
router = ShieldRouter(engine=engine)
@router.get("/orders")
async def get_orders():
"""Returns 200 normally; 503 during the scheduled maintenance window."""
return {"orders": [{"id": 1, "total": 49.99}, {"id": 2, "total": 129.00}]}
@router.get("/admin/schedule")
@force_active
async def schedule_maintenance():
"""Schedule a 10-second maintenance window starting 5 seconds from now."""
now = datetime.now(UTC)
window = MaintenanceWindow(
start=now + timedelta(seconds=5),
end=now + timedelta(seconds=15),
reason="Automated order system upgrade",
)
await engine.schedule_maintenance("GET:/orders", window=window, actor="demo")
return {
"scheduled": True,
"start": window.start.isoformat(),
"end": window.end.isoformat(),
"message": "GET /orders will enter maintenance in 5 seconds for 10 seconds",
}
@router.get("/admin/status")
@force_active
async def admin_status():
"""Current shield state for all registered routes."""
states = await engine.list_states()
return {"routes": [{"path": s.path, "status": s.status, "reason": s.reason} for s in states]}
@router.get("/health")
@force_active
async def health():
return {"status": "ok"}
# ---------------------------------------------------------------------------
# App assembly — scheduler is started inside ShieldEngine automatically
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(_: FastAPI):
await engine.start_scheduler()
yield
await engine.stop_scheduler()
app = FastAPI(
title="api-shield — Scheduled Maintenance Example",
description=(
"Hit `/admin/schedule` to trigger a 10-second maintenance window on "
"`GET /orders`. The window activates and deactivates automatically."
),
lifespan=lifespan,
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
Global maintenance¶
Blocking all routes at once
Demonstrates enabling and disabling global maintenance mode, which blocks every route in one call without per-route decorators. @force_active routes are exempt by default.
Endpoints:
| Endpoint | Purpose |
|---|---|
GET /payments |
Normal business route — blocked during global maintenance |
GET /orders |
Normal business route — blocked during global maintenance |
GET /health |
Always 200 (@force_active bypasses global maintenance) |
GET /admin/on |
Enable global maintenance |
GET /admin/off |
Disable global maintenance |
GET /admin/status |
Current global maintenance config |
Run:
Quick demo:
curl http://localhost:8000/payments # → 200
curl http://localhost:8000/admin/on # enable global maintenance
curl http://localhost:8000/payments # → 503 MAINTENANCE_MODE
curl http://localhost:8000/health # → 200 (force_active, exempt)
curl http://localhost:8000/admin/off # restore normal operation
curl http://localhost:8000/payments # → 200
Full source:
examples/fastapi/global_maintenance.py
"""FastAPI — Global Maintenance Mode Example.
Demonstrates enabling and disabling global maintenance mode, which blocks
every route at once without per-route decorators.
Run:
uv run uvicorn examples.fastapi.global_maintenance:app --reload
Endpoints:
GET /payments — normal business route
GET /orders — normal business route
GET /health — always 200 (@force_active, exempt from global maintenance)
GET /admin/on — enable global maintenance
GET /admin/off — disable global maintenance
GET /admin/status — show global maintenance state
Quick demo:
1. GET /payments → 200
2. GET /admin/on → enables global maintenance
3. GET /payments → 503 MAINTENANCE_MODE
4. GET /health → 200 (force_active routes are exempt by default)
5. GET /admin/off → disables global maintenance
6. GET /payments → 200 again
"""
from fastapi import FastAPI
from shield.core.backends.memory import MemoryBackend
from shield.core.engine import ShieldEngine
from shield.fastapi import ShieldMiddleware, ShieldRouter, force_active
engine = ShieldEngine(backend=MemoryBackend())
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# Business routes
# ---------------------------------------------------------------------------
@router.get("/payments")
async def get_payments():
"""Blocked when global maintenance is enabled."""
return {"payments": [{"id": 1, "amount": 99.99}]}
@router.get("/orders")
async def get_orders():
"""Blocked when global maintenance is enabled."""
return {"orders": [{"id": 1, "total": 49.99}]}
@router.get("/health")
@force_active
async def health():
"""Always 200 — @force_active routes bypass global maintenance by default."""
return {"status": "ok"}
# ---------------------------------------------------------------------------
# Admin routes — force_active so they stay reachable during maintenance
# ---------------------------------------------------------------------------
@router.get("/admin/on")
@force_active
async def enable_global_maintenance():
"""Enable global maintenance for all routes (except force_active)."""
await engine.enable_global_maintenance(
reason="Emergency infrastructure patch",
exempt_paths=["/health"],
)
return {"global_maintenance": "enabled"}
@router.get("/admin/off")
@force_active
async def disable_global_maintenance():
"""Disable global maintenance — all routes resume their per-route state."""
await engine.disable_global_maintenance()
return {"global_maintenance": "disabled"}
@router.get("/admin/status")
@force_active
async def admin_status():
"""Current global maintenance configuration."""
cfg = await engine.get_global_maintenance()
return {
"enabled": cfg.enabled,
"reason": cfg.reason,
"exempt_paths": cfg.exempt_paths,
"include_force_active": cfg.include_force_active,
}
# ---------------------------------------------------------------------------
# App assembly
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Global Maintenance Example",
description=(
"Hit `/admin/on` to enable global maintenance mode. "
"All routes return 503 except `@force_active` ones. "
"Hit `/admin/off` to restore normal operation."
),
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
Custom responses¶
HTML pages, redirects, and branded JSON errors
Shows how to replace the default JSON error body with any Starlette response — HTML maintenance pages, redirects, plain text, or a different JSON shape — either per-route or as an app-wide default on the middleware.
Resolution order: per-route response= → global responses= default → built-in JSON.
Expected behavior:
| Endpoint | Response | How |
|---|---|---|
GET /payments |
HTML maintenance page | Per-route factory on @maintenance |
GET /orders |
302 redirect to /status |
Per-route lambda on @maintenance |
GET /legacy |
Plain text 503 | Per-route lambda on @disabled |
GET /inventory |
HTML from global default | No per-route factory; falls back to middleware |
GET /reports |
HTML from global async factory | No per-route factory; async fallback |
GET /status |
200 JSON | Redirect target (@force_active) |
GET /health |
200 JSON | @force_active |
Run:
uv run uvicorn examples.fastapi.custom_responses:app --reload
# Admin dashboard: http://localhost:8000/shield/ (admin / secret)
Full source:
examples/fastapi/custom_responses.py
"""FastAPI — Custom Responses Example.
Demonstrates two ways to override the default JSON error response:
1. Per-route — pass ``response=`` directly to the decorator
2. Global — pass ``responses=`` to ``ShieldMiddleware`` as the app-wide default
Resolution order: per-route ``response=`` → global default → built-in JSON.
Run:
uv run uvicorn examples.fastapi.custom_responses:app --reload
Then visit:
http://localhost:8000/docs — Swagger UI
http://localhost:8000/shield/ — admin dashboard (login: admin / secret)
Try each blocked route to see its custom response:
GET /payments → HTML maintenance page (per-route, 503)
GET /orders → redirect to /status (per-route, 302)
GET /inventory → global HTML default (no per-route factory, falls back)
GET /reports → global HTML default (async factory on the global default)
GET /legacy → custom plain text (per-route on @disabled, 503)
GET /status → always 200 (the redirect target)
GET /health → always 200 (@force_active)
"""
import os
from fastapi import FastAPI
from starlette.requests import Request
from starlette.responses import (
HTMLResponse,
PlainTextResponse,
RedirectResponse,
)
from shield.admin import ShieldAdmin
from shield.core.config import make_engine
from shield.fastapi import (
ShieldMiddleware,
ShieldRouter,
apply_shield_to_openapi,
disabled,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# Shared response factories
# ---------------------------------------------------------------------------
def maintenance_html(request: Request, exc: Exception) -> HTMLResponse:
"""Reusable branded maintenance page — passed as response= or used globally."""
reason = getattr(exc, "reason", "Temporarily unavailable")
retry_after = getattr(exc, "retry_after", None)
retry_line = f"<p>Expected back: <strong>{retry_after}</strong></p>" if retry_after else ""
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Down for Maintenance</title>
<style>
body {{ font-family: sans-serif; display: flex; flex-direction: column;
align-items: center; justify-content: center; min-height: 100vh;
margin: 0; background: #f9fafb; color: #111; }}
h1 {{ font-size: 2rem; margin-bottom: .5rem; }}
p {{ color: #6b7280; }}
.tag {{ background: #fef3c7; color: #92400e; padding: .25rem .75rem;
border-radius: 9999px; font-size: .85rem; font-weight: 600; }}
</style>
</head>
<body>
<span class="tag">Maintenance</span>
<h1>We'll be right back</h1>
<p>{reason}</p>
{retry_line}
<p>Check <a href="/status">our status page</a> for live updates.</p>
</body>
</html>"""
return HTMLResponse(html, status_code=503)
def disabled_html(request: Request, exc: Exception) -> HTMLResponse:
"""Global default for @disabled routes."""
reason = getattr(exc, "reason", "This page is no longer available")
return HTMLResponse(
f"<h1>This page is gone</h1><p>{reason}</p><p><small>{request.url.path}</small></p>",
status_code=503,
)
async def async_maintenance_html(request: Request, exc: Exception) -> HTMLResponse:
"""Async variant — useful when the response body requires an awaited call
(template rendering, database lookup, etc.)."""
reason = getattr(exc, "reason", "Unavailable")
# In a real app: html = await templates.render("maintenance.html", reason=reason)
html = f"<h1>Unavailable</h1><p>{reason}</p><a href='/status'>Status</a>"
return HTMLResponse(html, status_code=503)
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health() -> dict:
"""Always 200 — bypasses every shield check."""
return {"status": "ok"}
@router.get("/status")
@force_active
async def status_page() -> dict:
"""Public status endpoint — the redirect target for /orders."""
return {"operational": True, "message": "Some services are under maintenance."}
# --- Per-route custom responses ---
@router.get("/payments")
@maintenance(reason="Database migration — back at 04:00 UTC", response=maintenance_html)
async def get_payments() -> dict:
"""Per-route HTML response — overrides the global default for this route only."""
return {"payments": []}
@router.get("/orders")
@maintenance(
reason="Order service upgrade",
response=lambda *_: RedirectResponse(url="/status", status_code=302),
)
async def get_orders() -> dict:
"""Per-route redirect — sends users to /status instead of an error body."""
return {"orders": []}
@router.get("/legacy")
@disabled(
reason="Retired. Use /v2/orders instead.",
response=lambda req, exc: PlainTextResponse(f"Gone. {exc.reason}", status_code=503),
)
async def legacy() -> dict:
"""Per-route plain text on a @disabled route."""
return {}
# --- Routes that fall back to the global default ---
@router.get("/inventory")
@maintenance(reason="Stock sync in progress")
async def get_inventory() -> dict:
"""No per-route response= set — falls back to middleware responses["maintenance"]."""
return {"items": []}
@router.get("/reports")
@maintenance(reason="Report generation paused for system upgrade")
async def get_reports() -> dict:
"""Also falls back to the global default (async factory)."""
return {"reports": []}
# ---------------------------------------------------------------------------
# App assembly
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Custom Responses",
description=(
"Shows two ways to customise blocked-route responses:\n\n"
"**Per-route**: `@maintenance(response=my_factory)` — overrides for one route.\n\n"
"**Global default**: `ShieldMiddleware(responses={...})` — applies to all routes "
"that have no per-route factory.\n\n"
"Resolution order: per-route → global default → built-in JSON."
),
)
# Global defaults — apply to any route that does NOT have response= on its decorator.
# Per-route factories always win.
app.add_middleware(
ShieldMiddleware,
engine=engine,
responses={
"maintenance": async_maintenance_html, # async factory works here too
"disabled": lambda *args: HTMLResponse(
f"<h1>This page is gone</h1><p>{args[1].reason}</p>", status_code=503
),
# "env_gated": ... # omit to keep the default silent 404
},
)
app.include_router(router)
apply_shield_to_openapi(app, engine)
app.mount(
"/shield",
ShieldAdmin(engine=engine, auth=("admin", "secret"), prefix="/shield"),
)
Webhooks¶
HTTP notifications on every state change
Fully self-contained webhook demo: three receivers (generic JSON, Slack-formatted, and a custom payload) are mounted on the same app — no external service needed. Change a route state via the CLI or dashboard and watch the events appear at /webhook-log.
Webhooks are always registered on the engine that owns state mutations:
- Embedded mode — register on the engine before passing it to
ShieldAdmin - Shield Server mode — build the engine explicitly and register on it before passing to
ShieldAdmin; SDK service apps never fire webhooks
# Shield Server mode
from shield.core.engine import ShieldEngine
from shield.core.webhooks import SlackWebhookFormatter
from shield.admin.app import ShieldAdmin
engine = ShieldEngine(backend=RedisBackend(...))
engine.add_webhook("https://hooks.slack.com/...", formatter=SlackWebhookFormatter())
shield_app = ShieldAdmin(engine=engine, auth=("admin", "secret"))
Webhook receivers (all @force_active):
| Endpoint | Payload format |
|---|---|
POST /webhooks/generic |
Default JSON (default_formatter) |
POST /webhooks/slack |
Slack Incoming Webhook blocks (SlackWebhookFormatter) |
POST /webhooks/custom |
Bespoke minimal payload (custom formatter) |
Run:
uv run uvicorn examples.fastapi.webhooks:app --reload
# Webhook log: http://localhost:8000/webhook-log (auto-refreshes every 5 s)
# Admin: http://localhost:8000/shield/ (admin / secret)
Trigger events:
shield config set-url http://localhost:8000/shield
shield login admin # password: secret
shield disable GET:/payments --reason "hotfix"
shield enable GET:/payments
shield maintenance GET:/orders --reason "stock sync"
shield enable GET:/orders
Then open http://localhost:8000/webhook-log to see all three receivers fire for each state change.
Full source:
examples/fastapi/webhooks.py
"""FastAPI — Webhooks Example.
Demonstrates how api-shield fires HTTP webhooks on every route state change.
This example is fully self-contained: the webhook receivers are mounted on the
same FastAPI app, so no external service is needed. Change a route state via
the CLI or admin dashboard and watch the events appear at /webhook-log.
Run:
uv run uvicorn examples.fastapi.webhooks:app --reload
Then open two terminals:
Terminal 1 — watch incoming webhook events:
watch -n1 curl -s http://localhost:8000/webhook-log
Terminal 2 — trigger state changes:
shield config set-url http://localhost:8000/shield
shield login admin # password: secret
shield disable GET:/payments --reason "hotfix"
shield enable GET:/payments
shield maintenance GET:/orders --reason "stock sync"
shield enable GET:/orders
Three webhooks are registered on startup:
1. /webhooks/generic — raw default_formatter JSON payload
2. /webhooks/slack — SlackWebhookFormatter payload (Slack-shaped blocks)
3. /webhooks/custom — bespoke formatter defined in this file
Visit http://localhost:8000/docs to explore all endpoints.
"""
import os
from collections import deque
from datetime import UTC, datetime
from typing import Any
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from shield.admin import ShieldAdmin
from shield.core.config import make_engine
from shield.core.models import RouteState
from shield.core.webhooks import SlackWebhookFormatter, default_formatter
from shield.fastapi import (
ShieldMiddleware,
ShieldRouter,
apply_shield_to_openapi,
disabled,
force_active,
maintenance,
)
# ---------------------------------------------------------------------------
# Engine & router
# ---------------------------------------------------------------------------
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# In-memory log — stores the last 50 webhook events received
# ---------------------------------------------------------------------------
_webhook_log: deque[dict[str, Any]] = deque(maxlen=50)
# ---------------------------------------------------------------------------
# Custom webhook formatter — bespoke payload shape
# ---------------------------------------------------------------------------
def custom_formatter(event: str, path: str, state: RouteState) -> dict[str, Any]:
"""Minimal custom formatter — returns only the fields our consumer needs."""
return {
"source": "api-shield",
"event": event,
"route": path,
"status": state.status,
"reason": state.reason or None,
"at": datetime.now(UTC).isoformat(),
}
# ---------------------------------------------------------------------------
# Webhook receiver endpoints (registered BEFORE webhooks are added so the
# URLs are live when the engine first fires)
# ---------------------------------------------------------------------------
@router.post("/webhooks/generic", include_in_schema=False)
@force_active
async def recv_generic(request: Request) -> dict:
"""Receives the default_formatter JSON payload."""
body = await request.json()
_webhook_log.appendleft({"receiver": "generic", "payload": body})
return {"ok": True}
@router.post("/webhooks/slack", include_in_schema=False)
@force_active
async def recv_slack(request: Request) -> dict:
"""Receives the SlackWebhookFormatter payload."""
body = await request.json()
_webhook_log.appendleft({"receiver": "slack", "payload": body})
return {"ok": True}
@router.post("/webhooks/custom", include_in_schema=False)
@force_active
async def recv_custom(request: Request) -> dict:
"""Receives the custom_formatter payload."""
body = await request.json()
_webhook_log.appendleft({"receiver": "custom", "payload": body})
return {"ok": True}
# ---------------------------------------------------------------------------
# Webhook log viewer
# ---------------------------------------------------------------------------
@router.get("/webhook-log", include_in_schema=True)
@force_active
async def webhook_log() -> HTMLResponse:
"""Browse the last 50 received webhook events as an HTML page.
Refresh this page after triggering state changes via the CLI or dashboard.
"""
import json
td = "padding:.5rem 1rem;border-bottom:1px solid #e5e7eb"
if not _webhook_log:
rows = (
"<tr><td colspan='3' style='text-align:center;color:#6b7280'>"
"No events yet — change a route state to trigger a webhook."
"</td></tr>"
)
else:
rows = ""
for entry in _webhook_log:
receiver = entry["receiver"]
payload = entry["payload"]
attachments = payload.get("attachments", [{}])
event = payload.get("event", attachments[0].get("text", "—"))
at = payload.get("timestamp") or payload.get("at") or "—"
detail = json.dumps(payload, indent=2)
rows += (
f"<tr>"
f'<td style="{td};font-weight:600">{receiver}</td>'
f'<td style="{td};font-family:monospace;font-size:.85rem">'
f"{event}</td>"
f'<td style="{td}"><details>'
f'<summary style="cursor:pointer;color:#6b7280">{at}</summary>'
f'<pre style="font-size:.8rem;margin:.5rem 0 0">{detail}'
f"</pre></details></td></tr>"
)
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta http-equiv="refresh" content="5">
<title>Webhook Log — api-shield</title>
<style>
body {{ font-family: sans-serif; margin: 2rem auto; max-width: 960px; color: #111; }}
h1 {{ font-size: 1.5rem; }}
p {{ color: #6b7280; font-size: .9rem; }}
table {{ width: 100%; border-collapse: collapse; }}
th {{ text-align: left; padding: .5rem 1rem; background: #f3f4f6;
border-bottom: 2px solid #e5e7eb; font-size: .85rem; color: #374151; }}
</style>
</head>
<body>
<h1>Webhook Log</h1>
<p>Showing last {len(_webhook_log)} of up to 50 events. Auto-refreshes every 5 seconds.</p>
<table>
<thead><tr><th>Receiver</th><th>Event</th><th>Timestamp / Payload</th></tr></thead>
<tbody>{rows}</tbody>
</table>
</body>
</html>"""
return HTMLResponse(html)
# ---------------------------------------------------------------------------
# Sample shielded routes — change their state to trigger webhooks
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health() -> dict:
"""Always 200 — use this to verify the server is up."""
return {"status": "ok", "env": CURRENT_ENV}
@router.get("/payments")
@maintenance(reason="Scheduled DB migration — back at 04:00 UTC")
async def get_payments() -> dict:
"""Starts in maintenance. Enable via CLI: shield enable GET:/payments"""
return {"payments": []}
@router.get("/orders")
async def get_orders() -> dict:
"""Starts active. Disable or maintenance via CLI to trigger a webhook."""
return {"orders": []}
@router.get("/legacy")
@disabled(reason="Replaced by /v2/legacy")
async def legacy() -> dict:
"""Permanently disabled — enable it via CLI to fire an enable webhook."""
return {}
# ---------------------------------------------------------------------------
# App assembly
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Webhooks Example",
description=(
"Demonstrates webhook notifications on route state changes.\n\n"
"**Webhook receivers** (all `@force_active`):\n"
"- `POST /webhooks/generic` — default JSON payload\n"
"- `POST /webhooks/slack` — Slack Incoming Webhook format\n"
"- `POST /webhooks/custom` — bespoke minimal payload\n\n"
"**Webhook log**: `GET /webhook-log` — auto-refreshes every 5 seconds.\n\n"
"Trigger events via the CLI or admin dashboard and watch them appear."
),
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
apply_shield_to_openapi(app, engine)
# ---------------------------------------------------------------------------
# Register webhooks — all three point at this same app (self-contained)
# ---------------------------------------------------------------------------
BASE_URL = os.getenv("APP_BASE_URL", "http://localhost:8000")
engine.add_webhook(f"{BASE_URL}/webhooks/generic", formatter=default_formatter)
engine.add_webhook(f"{BASE_URL}/webhooks/slack", formatter=SlackWebhookFormatter())
engine.add_webhook(f"{BASE_URL}/webhooks/custom", formatter=custom_formatter)
# ---------------------------------------------------------------------------
# Mount the admin dashboard + REST API (required for the CLI)
# ---------------------------------------------------------------------------
app.mount(
"/shield",
ShieldAdmin(engine=engine, auth=("admin", "secret"), prefix="/shield"),
)
Rate limiting¶
Per-IP, per-user, tiered limits, and custom 429 responses
Demonstrates IP-based, user-based, and tiered rate limiting with a custom 429 response factory. Requires api-shield[rate-limit].
Expected behavior:
| Endpoint | Limit | Key |
|---|---|---|
GET /public/posts |
5/minute | IP |
GET /users/me |
20/minute | user |
GET /reports |
free: 5/min, pro: 30/min | user + tier |
GET /health |
unlimited | @force_active |
Run:
uv add "api-shield[all,rate-limit]"
uv run uvicorn examples.fastapi.rate_limiting:app --reload
# Admin dashboard: http://localhost:8000/shield/ (admin / secret)
# Rate limits tab: http://localhost:8000/shield/rate-limits
# Blocked log: http://localhost:8000/shield/blocked
CLI quick-start:
shield login admin
shield rl list
shield rl set GET:/public/posts 20/minute # raise limit live
shield rl reset GET:/public/posts # clear counters
shield rl hits # blocked requests log
Full source:
examples/fastapi/rate_limiting.py
"""FastAPI — Rate Limiting Example.
Demonstrates the full @rate_limit decorator API:
* Basic IP-based limiting
* Per-user and per-API-key strategies
* Global (shared) counters
* Algorithms: fixed_window, sliding_window, moving_window, token_bucket
* Burst allowance
* Tiered limits (free / pro / enterprise)
* Exempt IPs
* on_missing_key behaviour (EXEMPT, FALLBACK_IP, BLOCK)
* Maintenance mode short-circuits the rate limit check
* Runtime policy mutation via the CLI (persisted to backend)
Run:
uv run uvicorn examples.fastapi.rate_limiting:app --reload
Then exercise the endpoints:
curl http://localhost:8000/docs # Swagger UI
curl http://localhost:8000/public/posts # IP-limited: 10/minute
curl http://localhost:8000/users/me # per-user: 100/minute
curl -H "X-API-Key: mykey" \\
http://localhost:8000/data # per-API-key: 50/minute
curl http://localhost:8000/search # global counter: 5/minute
curl http://localhost:8000/burst # 5/minute + burst 3 = 8 total
Admin dashboard (login: admin / secret):
http://localhost:8000/shield/
CLI — view and mutate policies at runtime (no redeploy needed):
shield login admin
shield rl list
shield rl set GET:/public/posts 20/minute # raise the limit live
shield rl set POST:/data 10/second --algorithm fixed_window
shield rl hits # blocked requests log
shield rl reset GET:/public/posts # clear counters
shield rl delete GET:/public/posts # remove persisted override
"""
from __future__ import annotations
from fastapi import FastAPI, Request
from shield.admin import ShieldAdmin
from shield.core.config import make_engine
from shield.fastapi import ShieldMiddleware, ShieldRouter, apply_shield_to_openapi, maintenance
from shield.fastapi.decorators import rate_limit
engine = make_engine()
router = ShieldRouter(engine=engine)
# ---------------------------------------------------------------------------
# 1. Basic IP-based limiting (default key strategy)
# ---------------------------------------------------------------------------
@router.get("/public/posts")
@rate_limit("10/minute")
async def list_posts():
"""10 requests/minute per IP address.
Responses include:
X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Reset
Blocked requests return 429 with Retry-After header.
"""
return {"posts": ["hello", "world"]}
# ---------------------------------------------------------------------------
# 2. Algorithm variants
# ---------------------------------------------------------------------------
@router.get("/fixed")
@rate_limit("5/minute", algorithm="fixed_window")
async def fixed_window_route():
"""Fixed window: counter resets hard at the window boundary.
Allows bursts at the boundary (all 5 at 00:59 + all 5 at 01:00).
Use when simplicity matters more than smoothness.
"""
return {"algorithm": "fixed_window"}
@router.get("/sliding")
@rate_limit("5/minute", algorithm="sliding_window")
async def sliding_window_route():
"""Sliding window: smooths out boundary bursts.
No request can exceed 5 within any rolling 60-second period.
This is the default algorithm.
"""
return {"algorithm": "sliding_window"}
@router.get("/moving")
@rate_limit("5/minute", algorithm="moving_window")
async def moving_window_route():
"""Moving window: strictest — tracks every individual request timestamp."""
return {"algorithm": "moving_window"}
@router.get("/token-bucket")
@rate_limit("5/minute", algorithm="token_bucket")
async def token_bucket_route():
"""Token bucket: allows controlled bursts, smoothed average rate."""
return {"algorithm": "token_bucket"}
# ---------------------------------------------------------------------------
# 3. Burst allowance
# ---------------------------------------------------------------------------
@router.get("/burst")
@rate_limit("5/minute", burst=3)
async def burst_route():
"""5/minute base rate + 3 burst = 8 total requests before blocking.
Burst lets clients absorb a short spike without hitting 429 immediately.
"""
return {"base": "5/minute", "burst": 3}
# ---------------------------------------------------------------------------
# 4. Per-user limiting (requires auth middleware to set request.state.user_id)
# ---------------------------------------------------------------------------
@router.get("/users/me")
@rate_limit("100/minute", key="user")
async def get_current_user(request: Request):
"""100 requests/minute per authenticated user.
Unauthenticated requests (no request.state.user_id) are EXEMPT by default —
they pass through without consuming the rate limit counter.
Set on_missing_key="block" to reject unauthenticated callers instead.
Simulate a logged-in user by setting request.state.user_id in a middleware.
"""
user_id = getattr(request.state, "user_id", "anonymous")
return {"user_id": user_id}
@router.get("/users/strict")
@rate_limit("100/minute", key="user", on_missing_key="block")
async def get_user_strict(request: Request):
"""Same limit, but unauthenticated callers get 429 instead of being exempt."""
user_id = getattr(request.state, "user_id", "anonymous")
return {"user_id": user_id}
@router.get("/users/fallback")
@rate_limit("100/minute", key="user", on_missing_key="fallback_ip")
async def get_user_fallback(request: Request):
"""When user_id is absent, fall back to IP-based counting."""
user_id = getattr(request.state, "user_id", "anonymous")
return {"user_id": user_id}
# ---------------------------------------------------------------------------
# 5. Per-API-key limiting
# ---------------------------------------------------------------------------
@router.get("/data")
@rate_limit("50/minute", key="api_key")
async def get_data(request: Request):
"""50 requests/minute per X-API-Key header value.
When the header is absent, falls back to IP-based limiting (api_key default).
Send the key via: curl -H "X-API-Key: mykey" http://localhost:8000/data
"""
api_key = request.headers.get("X-API-Key", "none")
return {"api_key": api_key, "data": [1, 2, 3]}
# ---------------------------------------------------------------------------
# 6. Global (shared) counter — all callers share one bucket
# ---------------------------------------------------------------------------
@router.get("/search")
@rate_limit("5/minute", key="global")
async def search():
"""5 requests/minute total across ALL callers.
Useful for protecting expensive endpoints against aggregate load
regardless of who is making the requests.
"""
return {"results": []}
# ---------------------------------------------------------------------------
# 7. Tiered limits — different rates per user plan
# (requires middleware to set request.state.plan = "free" | "pro" | "enterprise")
# ---------------------------------------------------------------------------
@router.get("/reports")
@rate_limit(
# Pass a dict to activate tiered mode.
# Keys are the values of request.state.plan (default tier_resolver).
# Requests with an unrecognised or missing plan get the "free" tier.
{"free": "10/minute", "pro": "100/minute", "enterprise": "unlimited"},
key="user",
)
async def get_reports(request: Request):
"""Tiered rate limiting based on request.state.plan.
free → 10/minute
pro → 100/minute
enterprise → unlimited (never blocked)
Set the tier in a middleware or dependency:
request.state.plan = user.subscription_plan
"""
plan = getattr(request.state, "plan", "free")
return {"plan": plan, "reports": []}
# ---------------------------------------------------------------------------
# 8. Exempt IPs — internal services / health checks bypass the limit
# ---------------------------------------------------------------------------
@router.get("/internal/metrics")
@rate_limit(
"10/minute",
exempt_ips=["127.0.0.1", "10.0.0.0/8", "192.168.0.0/16"],
)
async def internal_metrics():
"""Rate-limited externally, but localhost and RFC1918 ranges are exempt.
Useful for monitoring agents and internal services that poll frequently.
"""
return {"metrics": {"requests": 42}}
# ---------------------------------------------------------------------------
# 9. Maintenance + rate limit interaction
# Maintenance check runs BEFORE rate limit — quota is never consumed
# ---------------------------------------------------------------------------
@router.get("/checkout")
@maintenance(reason="Payment processor upgrade — back in 30 minutes")
@rate_limit("20/minute")
async def checkout():
"""While in maintenance, every request gets 503, not 429.
The rate limit counter is never incremented during maintenance so the
quota is fully preserved when the route comes back online.
Try it:
shield maintenance /checkout --reason "upgrade" # put in maintenance
# hammer the endpoint — counters stay at 0
shield enable /checkout # restore
# full quota available immediately
"""
return {"checkout": "ok"}
@router.get("/health")
def health():
"""Always 200 — use this to verify the server is up."""
return {"status": "ok"}
# ---------------------------------------------------------------------------
# App assembly
# ---------------------------------------------------------------------------
app = FastAPI(
title="api-shield — Rate Limiting Example",
description=(
"Demonstrates `@rate_limit` with all key strategies, algorithms, "
"burst, tiers, exempt IPs, and runtime policy mutation via the CLI."
),
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
apply_shield_to_openapi(app, engine)
app.mount(
"/shield",
ShieldAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/shield",
),
)
Shield Server (single service)¶
Centralized Shield Server + one service via ShieldSDK
Demonstrates the centralized Shield Server architecture: one Shield Server process owns all route state, and one service app connects via ShieldSDK. State is enforced locally — zero per-request network overhead.
Two ASGI apps — run each in its own terminal:
# Shield Server (port 8001)
uv run uvicorn examples.fastapi.shield_server:shield_app --port 8001 --reload
# Service app (port 8000)
uv run uvicorn examples.fastapi.shield_server:service_app --port 8000 --reload
Then visit:
http://localhost:8001/— Shield dashboard (admin/secret)http://localhost:8000/docs— service Swagger UI
Expected behavior:
| Endpoint | Response | Why |
|---|---|---|
GET /health |
200 always | @force_active |
GET /api/payments |
503 MAINTENANCE_MODE |
starts in maintenance |
GET /api/orders |
200 | active on startup |
GET /api/legacy |
503 ROUTE_DISABLED |
@disabled |
GET /api/v1/products |
200 + deprecation headers | @deprecated |
SDK authentication options:
# Option 1 — Auto-login (recommended): SDK logs in on startup, no token management
sdk = ShieldSDK(
server_url="http://localhost:8001",
app_id="payments-service",
username="admin",
password="secret", # inject from env in production
)
# Option 2 — Pre-issued token
sdk = ShieldSDK(
server_url="http://localhost:8001",
app_id="payments-service",
token="<token-from-shield-login>",
)
# Option 3 — No auth on the Shield Server
sdk = ShieldSDK(server_url="http://localhost:8001", app_id="payments-service")
CLI — always targets the Shield Server:
shield config set-url http://localhost:8001
shield login admin # password: secret
shield status
shield enable /api/payments
shield disable /api/orders --reason "hotfix"
shield maintenance /api/payments --reason "DB migration"
shield audit
Full source:
examples/fastapi/shield_server.py
"""FastAPI — Shield Server Mode Example.
Demonstrates the centralized Shield Server architecture: a single Shield
Server process owns all route state, and one or more service apps connect
to it via ShieldSDK. State is enforced locally on every request with zero
network overhead — the SDK keeps an in-process cache synced over a
persistent SSE connection.
This file defines TWO separate ASGI apps. Run them in separate terminals:
App 1 — The Shield Server (port 8001):
uv run uvicorn examples.fastapi.shield_server:shield_app --port 8001 --reload
App 2 — The Service App (port 8000):
uv run uvicorn examples.fastapi.shield_server:service_app --port 8000 --reload
Then visit:
http://localhost:8001/ — Shield Server dashboard (admin / secret)
http://localhost:8001/audit — audit log (all services)
http://localhost:8000/docs — service Swagger UI
CLI — always points at the Shield Server, not the service:
shield config set-url http://localhost:8001
shield login admin # password: secret
shield status # routes registered by my-service
shield disable /api/orders --reason "hotfix"
shield enable /api/orders
shield maintenance /api/payments --reason "DB migration"
shield audit # full audit trail
Expected behaviour:
GET /health → 200 always (@force_active — survives disable)
GET /api/payments → 503 MAINTENANCE_MODE (starts in maintenance)
GET /api/orders → 200 (active on startup)
GET /api/legacy → 503 ROUTE_DISABLED (@disabled)
GET /api/v1/products → 200 + deprecation hdr (@deprecated)
GET /api/v2/products → 200 (active successor)
Production notes:
Backend choice affects the Shield Server only — SDK clients always receive
live SSE updates regardless of backend, because they connect to the Shield
Server over HTTP (not to the backend directly):
* MemoryBackend — fine for development; state is lost when the Shield
Server restarts.
* FileBackend — state survives restarts; safe for single-server
deployments (no multi-process file locking).
* RedisBackend — required only when you run multiple Shield Server
instances behind a load balancer (high availability).
Cross-instance pub/sub keeps all Shield Server nodes
in sync so every SDK client gets consistent state.
* Use a stable secret_key so tokens survive Shield Server restarts.
* Prefer passing username/password to ShieldSDK so the SDK obtains its
own sdk-platform token on startup (sdk_token_expiry, default 1 year)
rather than managing a pre-issued token manually.
* Set token_expiry (dashboard/CLI sessions) and sdk_token_expiry (service
tokens) independently so human sessions stay short-lived.
"""
from __future__ import annotations
from fastapi import FastAPI
from shield.core.backends.memory import MemoryBackend
from shield.fastapi import (
ShieldRouter,
apply_shield_to_openapi,
deprecated,
disabled,
force_active,
maintenance,
)
from shield.sdk import ShieldSDK
from shield.server import ShieldServer
# ---------------------------------------------------------------------------
# App 1 — Shield Server
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.shield_server:shield_app --port 8001 --reload
#
# The Shield Server is a self-contained ASGI app that exposes:
# / — HTMX dashboard UI (login: admin / secret)
# /audit — audit log
# /api/... — REST API consumed by the CLI
# /api/sdk/... — SSE + register endpoints consumed by ShieldSDK clients
#
# For production: swap MemoryBackend for RedisBackend so every connected
# service receives live state updates via the SSE channel.
#
# from shield.core.backends.redis import RedisBackend
# backend = RedisBackend("redis://localhost:6379")
#
# secret_key should be a stable value so issued tokens survive restarts.
# Omit it (or pass None) in development — a random key is generated per run.
shield_app = ShieldServer(
backend=MemoryBackend(),
auth=("admin", "secret"),
# secret_key="change-me-in-production",
# token_expiry=3600, # dashboard / CLI sessions — default 24 h
# sdk_token_expiry=31536000, # SDK service tokens — default 1 year
)
# ---------------------------------------------------------------------------
# App 2 — Service App
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.shield_server:service_app --port 8000 --reload
#
# ShieldSDK wires ShieldMiddleware + startup/shutdown lifecycle into the app.
# Route enforcement is purely local — the SDK never adds per-request latency.
#
# Authentication options (choose one):
#
# 1. Auto-login — recommended for production. The SDK calls
# POST /api/auth/login with platform="sdk" on startup and caches the
# returned token (valid for sdk_token_expiry, default 1 year).
# Inject credentials from environment variables:
#
# sdk = ShieldSDK(
# server_url="http://localhost:8001",
# app_id="my-service",
# username=os.environ["SHIELD_USERNAME"],
# password=os.environ["SHIELD_PASSWORD"],
# )
#
# 2. Pre-issued token — obtain once via `shield login`, store as a secret:
# sdk = ShieldSDK(..., token=os.environ["SHIELD_TOKEN"])
#
# 3. No auth — omit token/username/password when the Shield Server has
# no auth configured (auth=None or auth omitted).
sdk = ShieldSDK(
server_url="http://localhost:8001",
app_id="my-service",
username="admin",
password="secret",
# username="admin", # or inject from env: os.environ["SHIELD_USERNAME"]
# password="secret", # or inject from env: os.environ["SHIELD_PASSWORD"]
reconnect_delay=5.0, # seconds between SSE reconnect attempts
)
service_app = FastAPI(
title="api-shield — Shield Server Example (Service)",
description=(
"Connects to the Shield Server at **http://localhost:8001** via "
"ShieldSDK. All route state is managed centrally — use the "
"[Shield Dashboard](http://localhost:8001/) or the CLI to "
"enable, disable, or pause any route without redeploying."
),
)
# attach() adds ShieldMiddleware and wires startup/shutdown hooks.
# Call this BEFORE defining routes so the router below can use sdk.engine.
sdk.attach(service_app)
# ShieldRouter auto-registers decorated routes with the Shield Server on
# startup so they appear in the dashboard immediately.
router = ShieldRouter(engine=sdk.engine)
# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health():
"""Always 200 — bypasses every shield check.
Use this for load-balancer probes. @force_active ensures the route
stays reachable even if the Shield Server is temporarily unreachable
and the SDK falls back to its empty local cache.
"""
return {"status": "ok", "service": "my-service"}
@router.get("/api/payments")
@maintenance(reason="Scheduled database migration — back at 04:00 UTC")
async def get_payments():
"""Returns 503 MAINTENANCE_MODE on startup.
Lift maintenance from the CLI (no redeploy needed):
shield enable /api/payments
"""
return {"payments": [{"id": 1, "amount": 99.99}]}
@router.get("/api/orders")
async def list_orders():
"""Active on startup — disable from the CLI:
shield disable /api/orders --reason "hotfix"
shield enable /api/orders
"""
return {"orders": [{"id": 42, "status": "shipped"}]}
@router.get("/api/legacy")
@disabled(reason="Use /api/v2/products instead")
async def legacy_endpoint():
"""Returns 503 ROUTE_DISABLED.
The @disabled state is set at deploy time and can be overridden from
the dashboard or CLI:
shield enable /api/legacy
"""
return {}
@router.get("/api/v1/products")
@deprecated(sunset="Sat, 01 Jan 2028 00:00:00 GMT", use_instead="/api/v2/products")
async def v1_products():
"""Returns 200 with Deprecation, Sunset, and Link response headers.
Headers injected by ShieldMiddleware on every response:
Deprecation: true
Sunset: Sat, 01 Jan 2028 00:00:00 GMT
Link: </api/v2/products>; rel="successor-version"
"""
return {"products": [{"id": 1, "name": "Widget"}], "version": 1}
@router.get("/api/v2/products")
async def v2_products():
"""Active successor to /api/v1/products."""
return {"products": [{"id": 1, "name": "Widget"}], "version": 2}
service_app.include_router(router)
apply_shield_to_openapi(service_app, sdk.engine)
# ---------------------------------------------------------------------------
# How the CLI talks to this setup
# ---------------------------------------------------------------------------
#
# The CLI always communicates with the Shield Server, never directly with
# the service app. From the Shield Server's perspective, routes from
# "my-service" appear namespaced as "my-service:/api/payments" etc.
#
# # One-time setup
# shield config set-url http://localhost:8001
# shield login admin
#
# # Inspect state
# shield status # all routes for my-service
# shield audit # full audit trail
#
# # Lifecycle management
# shield disable /api/orders --reason "hotfix"
# shield enable /api/orders
# shield maintenance /api/payments --reason "scheduled downtime"
# shield schedule /api/payments # set maintenance window
#
# # Dashboard
# open http://localhost:8001/ # full UI, no CLI needed
Shield Server (multi-service)¶
Two independent services sharing one Shield Server
Demonstrates two independent FastAPI services (payments-service and orders-service) both connecting to the same Shield Server. Each service registers its routes under its own app_id namespace so the dashboard service dropdown and CLI SHIELD_SERVICE env var can manage them independently or together.
Each service authenticates using username/password so the SDK obtains its own long-lived sdk-platform token on startup — no manual token management required. The Shield Server is configured with separate expiry times for human sessions and service tokens:
shield_app = ShieldServer(
backend=MemoryBackend(),
auth=("admin", "secret"),
token_expiry=3600, # dashboard / CLI: 1 hour
sdk_token_expiry=31536000, # SDK services: 1 year
)
payments_sdk = ShieldSDK(
server_url="http://shield-server:9000",
app_id="payments-service",
username="admin",
password="secret", # inject from env in production
)
Three ASGI apps — run each in its own terminal:
# Shield Server (port 8001)
uv run uvicorn examples.fastapi.multi_service:shield_app --port 8001 --reload
# Payments service (port 8000)
uv run uvicorn examples.fastapi.multi_service:payments_app --port 8000 --reload
# Orders service (port 8002)
uv run uvicorn examples.fastapi.multi_service:orders_app --port 8002 --reload
Then visit:
http://localhost:8001/— Shield dashboard (use service dropdown to switch)http://localhost:8000/docs— Payments Swagger UIhttp://localhost:8002/docs— Orders Swagger UI
Expected behavior:
| Service | Endpoint | Response | Why |
|---|---|---|---|
| payments | GET /health |
200 always | @force_active |
| payments | GET /api/payments |
503 MAINTENANCE_MODE |
starts in maintenance |
| payments | GET /api/refunds |
200 | active |
| payments | GET /api/v1/invoices |
200 + deprecation headers | @deprecated |
| orders | GET /health |
200 always | @force_active |
| orders | GET /api/orders |
200 | active |
| orders | GET /api/shipments |
503 ROUTE_DISABLED |
@disabled |
| orders | GET /api/cart |
200 | active |
CLI — multi-service workflow:
shield config set-url http://localhost:8001
shield login admin # password: secret
shield services # list all connected services
# Scope to payments via env var
export SHIELD_SERVICE=payments-service
shield status
shield enable /api/payments
shield current-service # confirm active context
# Switch to orders with explicit flag (overrides env var)
shield status --service orders-service
shield disable /api/cart --reason "redesign" --service orders-service
# Unscoped — operates across all services
unset SHIELD_SERVICE
shield status
shield audit
shield global disable --reason "emergency maintenance"
shield global enable
Full source:
examples/fastapi/multi_service.py
"""FastAPI — Multi-Service Shield Server Example.
Demonstrates two independent FastAPI services (payments and orders) both
connecting to the same Shield Server. Each service registers its routes
under its own app_id namespace so the dashboard and CLI can manage them
independently or together.
This file defines THREE separate ASGI apps. Run each in its own terminal:
Shield Server (port 8001):
uv run --with uvicorn uvicorn examples.fastapi.multi_service:shield_app --port 8001 --reload
Payments service (port 8000):
uv run --with uvicorn uvicorn examples.fastapi.multi_service:payments_app --port 8000 --reload
Orders service (port 8002):
uv run --with uvicorn uvicorn examples.fastapi.multi_service:orders_app --port 8002 --reload
Then visit:
http://localhost:8001/ — Shield dashboard (admin / secret)
Use the service dropdown to switch between
"payments-service" and "orders-service"
http://localhost:8000/docs — Payments Swagger UI
http://localhost:8002/docs — Orders Swagger UI
CLI — points at the Shield Server; use --service or SHIELD_SERVICE to scope:
# One-time setup
shield config set-url http://localhost:8001
shield login admin # password: secret
# View all registered services
shield services
# Manage payments routes
export SHIELD_SERVICE=payments-service
shield status
shield disable /api/payments --reason "hotfix"
shield enable /api/payments
# Switch to orders without changing env var
shield status --service orders-service
# Explicit --service flag overrides the env var
export SHIELD_SERVICE=payments-service
shield enable /api/orders --service orders-service
# Clear the env var to work across all services at once
unset SHIELD_SERVICE
shield status # shows routes from both services
shield audit # audit log from both services
# Global maintenance — affects ALL services
shield global disable --reason "emergency maintenance"
shield global enable
Expected behaviour:
Payments (port 8000):
GET /health → 200 always (@force_active)
GET /api/payments → 503 MAINTENANCE_MODE (starts in maintenance)
GET /api/refunds → 200 (active)
GET /api/v1/invoices → 200 + deprecation hdr (@deprecated)
GET /api/v2/invoices → 200 (active successor)
Orders (port 8002):
GET /health → 200 always (@force_active)
GET /api/orders → 200 (active)
GET /api/shipments → 503 ROUTE_DISABLED (@disabled)
GET /api/cart → 200 (active)
Production notes:
Backend choice affects the Shield Server only. All SDK clients receive
live SSE updates regardless of backend — they connect to the Shield Server
over HTTP, never to the backend directly:
* MemoryBackend — fine for development; state lost on Shield Server restart.
* FileBackend — state survives restarts; single Shield Server instance only.
* RedisBackend — needed only when running multiple Shield Server instances
(HA / load-balanced). Redis pub/sub keeps all nodes in
sync so every SDK client sees consistent state.
* Use a stable secret_key so tokens survive Shield Server restarts.
* Prefer passing username/password to each ShieldSDK so each service
obtains its own sdk-platform token on startup automatically.
* Set token_expiry (dashboard/CLI) and sdk_token_expiry (services)
independently so human sessions stay short-lived.
"""
from __future__ import annotations
from fastapi import FastAPI
from shield.core.backends.memory import MemoryBackend
from shield.fastapi import (
ShieldRouter,
apply_shield_to_openapi,
deprecated,
disabled,
force_active,
maintenance,
setup_shield_docs,
)
from shield.sdk import ShieldSDK
from shield.server import ShieldServer
# ---------------------------------------------------------------------------
# Shield Server — shared by all services
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.multi_service:shield_app --port 8001 --reload
#
# All services register their routes here. The dashboard service dropdown
# lets you filter and manage each service independently.
#
# For production: swap MemoryBackend for RedisBackend:
# from shield.core.backends.redis import RedisBackend
# backend = RedisBackend("redis://localhost:6379")
shield_app = ShieldServer(
backend=MemoryBackend(),
auth=("admin", "secret"),
# secret_key="change-me-in-production",
# token_expiry=3600, # dashboard / CLI sessions — default 24 h
# sdk_token_expiry=31536000, # SDK service tokens — default 1 year
)
# ---------------------------------------------------------------------------
# Payments Service (port 8000)
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.multi_service:payments_app --port 8000 --reload
#
# app_id="payments-service" namespaces all routes from this service on the
# Shield Server. The dashboard shows them separately from orders-service.
# CLI: export SHIELD_SERVICE=payments-service; shield status
payments_sdk = ShieldSDK(
server_url="http://localhost:8001",
app_id="payments-service",
username="admin",
password="secret",
# Auto-login (recommended): SDK obtains a 1-year sdk-platform token on startup.
# username="admin", # inject from env: os.environ["SHIELD_USERNAME"]
# password="secret", # inject from env: os.environ["SHIELD_PASSWORD"]
# Or use a pre-issued token: token=os.environ["SHIELD_TOKEN"]
reconnect_delay=5.0,
)
payments_app = FastAPI(
title="api-shield — Payments Service",
description=(
"Connects to the Shield Server at **http://localhost:8001** as "
"`payments-service`. Manage routes from the "
"[Shield Dashboard](http://localhost:8001/) or via the CLI with "
"`export SHIELD_SERVICE=payments-service`."
),
)
payments_sdk.attach(payments_app)
payments_router = ShieldRouter(engine=payments_sdk.engine)
@payments_router.get("/health")
@force_active
async def payments_health():
"""Always 200 — load-balancer probe endpoint."""
return {"status": "ok", "service": "payments-service"}
@payments_router.get("/api/payments")
@maintenance(reason="Payment processor upgrade — back at 04:00 UTC")
async def process_payment():
"""Returns 503 MAINTENANCE_MODE on startup.
Restore from the CLI:
export SHIELD_SERVICE=payments-service
shield enable /api/payments
"""
return {"payment_id": "pay_abc123", "status": "processed"}
@payments_router.get("/api/refunds")
async def list_refunds():
"""Active on startup.
Disable from the CLI:
shield disable /api/refunds --reason "audit in progress" \\
--service payments-service
"""
return {"refunds": [{"id": "ref_001", "amount": 49.99}]}
@payments_router.get("/api/v1/invoices")
@deprecated(sunset="Sat, 01 Jun 2028 00:00:00 GMT", use_instead="/api/v2/invoices")
async def v1_invoices():
"""Returns 200 with Deprecation, Sunset, and Link response headers."""
return {"invoices": [{"id": "inv_001", "total": 199.99}], "version": 1}
@payments_router.get("/api/v2/invoices")
async def v2_invoices():
"""Active successor to /api/v1/invoices."""
return {"invoices": [{"id": "inv_001", "total": 199.99}], "version": 2}
payments_app.include_router(payments_router)
apply_shield_to_openapi(payments_app, payments_sdk.engine)
setup_shield_docs(payments_app, payments_sdk.engine)
# ---------------------------------------------------------------------------
# Orders Service (port 8002)
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.multi_service:orders_app --port 8002 --reload
#
# app_id="orders-service" gives this service its own namespace on the server.
# CLI: export SHIELD_SERVICE=orders-service; shield status
orders_sdk = ShieldSDK(
server_url="http://localhost:8001",
app_id="orders-service",
username="admin",
password="secret",
# Auto-login (recommended): SDK obtains a 1-year sdk-platform token on startup.
# username="admin", # inject from env: os.environ["SHIELD_USERNAME"]
# password="secret", # inject from env: os.environ["SHIELD_PASSWORD"]
# Or use a pre-issued token: token=os.environ["SHIELD_TOKEN"]
reconnect_delay=5.0,
)
orders_app = FastAPI(
title="api-shield — Orders Service",
description=(
"Connects to the Shield Server at **http://localhost:8001** as "
"`orders-service`. Manage routes from the "
"[Shield Dashboard](http://localhost:8001/) or via the CLI with "
"`export SHIELD_SERVICE=orders-service`."
),
)
orders_sdk.attach(orders_app)
orders_router = ShieldRouter(engine=orders_sdk.engine)
@orders_router.get("/health")
@force_active
async def orders_health():
"""Always 200 — load-balancer probe endpoint."""
return {"status": "ok", "service": "orders-service"}
@orders_router.get("/api/orders")
async def list_orders():
"""Active on startup.
Disable from the CLI:
shield disable /api/orders --reason "inventory sync" \\
--service orders-service
"""
return {"orders": [{"id": 42, "status": "shipped"}]}
@orders_router.get("/api/shipments")
@disabled(reason="Shipment provider integration deprecated — use /api/orders")
async def list_shipments():
"""Returns 503 ROUTE_DISABLED.
Re-enable from the CLI if you need to temporarily restore access:
shield enable /api/shipments --service orders-service
"""
return {}
@orders_router.get("/api/cart")
async def get_cart():
"""Active on startup.
Put the whole orders-service in global maintenance from the dashboard
or pause just this route:
shield maintenance /api/cart --reason "cart redesign" \\
--service orders-service
"""
return {"cart": {"items": [], "total": 0.0}}
orders_app.include_router(orders_router)
apply_shield_to_openapi(orders_app, orders_sdk.engine)
setup_shield_docs(orders_app, orders_sdk.engine)
# ---------------------------------------------------------------------------
# CLI reference — multi-service workflow
# ---------------------------------------------------------------------------
#
# Setup (once):
# shield config set-url http://localhost:8001
# shield login admin
#
# View all services and their routes:
# shield services
# shield status # routes from ALL services combined
#
# Scope to a specific service via env var:
# export SHIELD_SERVICE=payments-service
# shield status # only payments-service routes
# shield disable /api/payments --reason "hotfix"
# shield enable /api/payments
# shield maintenance /api/refunds --reason "audit"
#
# Switch service without changing the env var (--service flag):
# shield status --service orders-service
# shield disable /api/orders --reason "inventory sync" \\
# --service orders-service
#
# Explicit flag always overrides the SHIELD_SERVICE env var:
# export SHIELD_SERVICE=payments-service
# shield enable /api/orders --service orders-service # acts on orders
#
# Unscoped commands operate across all services:
# unset SHIELD_SERVICE
# shield audit # audit log from both services
# shield global disable --reason "emergency maintenance"
# shield global enable
Custom backend (SQLite)¶
Implementing ShieldBackend with aiosqlite
A complete, working custom backend that stores all route state and audit log entries in a SQLite database via aiosqlite. Restart the server and the state survives. The same CLI workflow works unchanged — the CLI talks to the app's REST API, never to the database directly.
Requirements:
Expected behavior:
| Endpoint | Response |
|---|---|
GET /health |
200 always — backend: "sqlite" in body |
GET /payments |
503 MAINTENANCE_MODE (persisted in SQLite) |
GET /legacy |
503 ROUTE_DISABLED (persisted in SQLite) |
GET /orders |
200 active |
Run:
uv run uvicorn examples.fastapi.custom_backend.sqlite_backend:app --reload
# Swagger UI: http://localhost:8000/docs
# Admin: http://localhost:8000/shield/ (admin / secret)
# Audit log: http://localhost:8000/shield/audit
CLI quick-start:
shield config set-url http://localhost:8000/shield
shield login admin # password: secret
shield status
shield disable GET:/payments --reason "hotfix"
shield enable GET:/payments
shield log
Full source:
examples/fastapi/custom_backend/sqlite_backend.py
"""Custom Backend Example — SQLite via aiosqlite.
This file shows how to wire api-shield to a storage layer it does not ship
with by implementing the ``ShieldBackend`` abstract base class.
The contract is simple: implement six async methods and api-shield handles the
rest (engine logic, middleware, decorators, dashboard, CLI, audit log).
Requirements:
pip install aiosqlite
# or: uv add aiosqlite
Run the demo app:
uv run uvicorn examples.fastapi.custom_backend.sqlite_backend:app --reload
Then visit:
http://localhost:8000/docs — filtered Swagger UI
http://localhost:8000/shield/ — admin dashboard (login: admin / secret)
http://localhost:8000/shield/audit — audit log
CLI quick-start (the CLI talks to the app's admin API — it never touches
the database directly):
shield config set-url http://localhost:8000/shield
shield login admin # password: secret
shield status
shield disable GET:/payments --reason "hotfix"
shield enable GET:/payments
shield log
"""
from __future__ import annotations
import sqlite3
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
import aiosqlite
from fastapi import FastAPI
from shield.admin import ShieldAdmin
from shield.core.backends.base import ShieldBackend
from shield.core.engine import ShieldEngine
from shield.core.models import AuditEntry, RouteState
from shield.fastapi import (
ShieldMiddleware,
ShieldRouter,
apply_shield_to_openapi,
disabled,
force_active,
maintenance,
)
# ---------------------------------------------------------------------------
# SQLiteBackend — implements the ShieldBackend contract
#
# Rules to follow when building any custom backend:
#
# 1. Subclass ``ShieldBackend`` from ``shield.core.backends.base``.
# 2. Implement all six @abstractmethod methods.
# 3. Override ``startup()`` / ``shutdown()`` for async initialisation —
# the engine calls these automatically when used as ``async with engine:``.
# 4. RouteState and AuditEntry are Pydantic models — use .model_dump_json()
# to serialise and .model_validate_json() to deserialise.
# 5. get_state() must raise KeyError when the path is not found.
# 6. Fail-open: let exceptions bubble up — ShieldEngine wraps every backend
# call in try/except and allows the request through on failure.
# 7. subscribe() is optional. Leave it as-is if your backend doesn't support
# pub/sub (the base class raises NotImplementedError and the dashboard
# falls back to polling).
# ---------------------------------------------------------------------------
_MAX_AUDIT_ROWS = 1000
_CREATE_STATES_TABLE = """
CREATE TABLE IF NOT EXISTS shield_states (
path TEXT PRIMARY KEY,
state_json TEXT NOT NULL
)
"""
_CREATE_AUDIT_TABLE = """
CREATE TABLE IF NOT EXISTS shield_audit (
id TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
path TEXT NOT NULL,
entry_json TEXT NOT NULL
)
"""
class SQLiteBackend(ShieldBackend):
"""api-shield backend backed by a SQLite database.
Parameters
----------
db_path:
Path to the SQLite file. Use ``:memory:`` for an in-process
database (useful for tests — not shared across processes).
Example
-------
>>> backend = SQLiteBackend("shield-state.db")
>>> engine = ShieldEngine(backend=backend)
>>> async with engine: # calls startup() then shutdown()
... states = await engine.list_states()
"""
def __init__(self, db_path: str = "shield-state.db") -> None:
self._db_path = db_path
self._db: aiosqlite.Connection | None = None
# ------------------------------------------------------------------
# Lifecycle hooks — called automatically by ShieldEngine
# ------------------------------------------------------------------
async def startup(self) -> None:
"""Open the database connection and create tables if needed.
Called automatically by ``ShieldEngine.__aenter__``. You do not
need to call this yourself when using ``async with engine:``.
"""
self._db = await aiosqlite.connect(self._db_path)
self._db.row_factory = sqlite3.Row
await self._db.execute(_CREATE_STATES_TABLE)
await self._db.execute(_CREATE_AUDIT_TABLE)
await self._db.commit()
async def shutdown(self) -> None:
"""Close the database connection.
Called automatically by ``ShieldEngine.__aexit__``.
"""
if self._db is not None:
await self._db.close()
self._db = None
@property
def _conn(self) -> aiosqlite.Connection:
if self._db is None:
raise RuntimeError(
"SQLiteBackend is not connected. "
"Use 'async with engine:' to ensure startup() is called."
)
return self._db
# ------------------------------------------------------------------
# ShieldBackend — required methods
# ------------------------------------------------------------------
async def get_state(self, path: str) -> RouteState:
"""Return the stored state for *path*.
Raises ``KeyError`` if the path has not been registered yet —
this is the contract the engine relies on to distinguish
"not registered" from "registered but active".
"""
async with self._conn.execute(
"SELECT state_json FROM shield_states WHERE path = ?", (path,)
) as cursor:
row = await cursor.fetchone()
if row is None:
raise KeyError(f"No state registered for path {path!r}")
return RouteState.model_validate_json(row["state_json"])
async def set_state(self, path: str, state: RouteState) -> None:
"""Persist *state* for *path*, creating or replacing the existing row."""
await self._conn.execute(
"""
INSERT INTO shield_states (path, state_json)
VALUES (?, ?)
ON CONFLICT(path) DO UPDATE SET state_json = excluded.state_json
""",
(path, state.model_dump_json()),
)
await self._conn.commit()
async def delete_state(self, path: str) -> None:
"""Remove the state row for *path*. No-op if not found."""
await self._conn.execute("DELETE FROM shield_states WHERE path = ?", (path,))
await self._conn.commit()
async def list_states(self) -> list[RouteState]:
"""Return all registered route states."""
async with self._conn.execute("SELECT state_json FROM shield_states") as cursor:
rows = await cursor.fetchall()
return [RouteState.model_validate_json(row["state_json"]) for row in rows]
async def write_audit(self, entry: AuditEntry) -> None:
"""Append *entry* to the audit log, capping the table at 1000 rows."""
await self._conn.execute(
"""
INSERT OR IGNORE INTO shield_audit (id, timestamp, path, entry_json)
VALUES (?, ?, ?, ?)
""",
(
entry.id,
entry.timestamp.isoformat(),
entry.path,
entry.model_dump_json(),
),
)
# Keep the table from growing unbounded.
await self._conn.execute(
"""
DELETE FROM shield_audit
WHERE id NOT IN (
SELECT id FROM shield_audit
ORDER BY timestamp DESC
LIMIT ?
)
""",
(_MAX_AUDIT_ROWS,),
)
await self._conn.commit()
async def get_audit_log(self, path: str | None = None, limit: int = 100) -> list[AuditEntry]:
"""Return audit entries, newest first, optionally filtered by *path*."""
if path is not None:
query = """
SELECT entry_json FROM shield_audit
WHERE path = ?
ORDER BY timestamp DESC
LIMIT ?
"""
params: tuple[object, ...] = (path, limit)
else:
query = """
SELECT entry_json FROM shield_audit
ORDER BY timestamp DESC
LIMIT ?
"""
params = (limit,)
async with self._conn.execute(query, params) as cursor:
rows = await cursor.fetchall()
return [AuditEntry.model_validate_json(row["entry_json"]) for row in rows]
# ------------------------------------------------------------------
# subscribe() — not implemented; dashboard falls back to polling
# ------------------------------------------------------------------
async def subscribe(self) -> AsyncIterator[RouteState]: # type: ignore[return]
raise NotImplementedError(
"SQLiteBackend does not support pub/sub. The dashboard will use polling instead."
)
yield # makes the type checker treat this as an async generator
# ---------------------------------------------------------------------------
# Demo FastAPI app using SQLiteBackend
# ---------------------------------------------------------------------------
backend = SQLiteBackend("shield-state.db")
engine = ShieldEngine(backend=backend)
router = ShieldRouter(engine=engine)
@router.get("/health")
@force_active
async def health():
"""Always 200 — bypasses every shield check."""
return {"status": "ok", "backend": "sqlite"}
@router.get("/payments")
@maintenance(reason="DB migration — back at 04:00 UTC")
async def get_payments():
"""Returns 503 MAINTENANCE_MODE — state persisted in SQLite."""
return {"payments": []}
@router.get("/legacy")
@disabled(reason="Use /payments instead")
async def legacy():
"""Returns 503 ROUTE_DISABLED — state persisted in SQLite."""
return {}
@router.get("/orders")
async def get_orders():
"""200 active — no decorator, state persists across restarts via SQLite."""
return {"orders": [{"id": 1, "total": 49.99}]}
@asynccontextmanager
async def lifespan(_: FastAPI):
# async with engine: calls backend.startup() then backend.shutdown()
async with engine:
yield
app = FastAPI(
title="api-shield — SQLite Custom Backend Example",
description=(
"All route state and audit log entries are persisted in `shield-state.db`. "
"Restart the server and the state survives.\n\n"
"Admin UI and CLI API available at `/shield/`."
),
lifespan=lifespan,
)
app.add_middleware(ShieldMiddleware, engine=engine)
app.include_router(router)
apply_shield_to_openapi(app, engine)
# Mount the unified admin interface:
# - Dashboard UI → http://localhost:8000/shield/
# - REST API → http://localhost:8000/shield/api/... (used by the CLI)
#
# The CLI communicates with the app via this REST API — it never touches
# the SQLite database directly. This means the same CLI workflow works
# regardless of which backend the app uses.
app.mount(
"/shield",
ShieldAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/shield",
# secret_key="change-me-in-production",
# token_expiry=86400, # seconds — default 24 h
),
)