FastAPI Adapter¶
The FastAPI adapter provides middleware, decorators, a drop-in router, and OpenAPI integration — all built on top of the framework-agnostic waygate.core.
More adapters on the way
We currently support FastAPI. Other framework adapters are on the way. Open an issue if you'd like to see your framework supported sooner.
Installation¶
uv add "waygate[fastapi]" # adapter only
uv add "waygate[fastapi,rate-limit]" # with rate limiting
uv add "waygate[all]" # everything including CLI + admin
Quick setup¶
main.py
from fastapi import FastAPI
from waygate import make_engine
from waygate.fastapi import (
WaygateMiddleware,
WaygateAdmin,
apply_waygate_to_openapi,
setup_waygate_docs,
maintenance,
env_only,
disabled,
force_active,
deprecated,
)
engine = make_engine() # reads WAYGATE_BACKEND, WAYGATE_ENV from env / .waygate
app = FastAPI()
app.add_middleware(WaygateMiddleware, 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_waygate_to_openapi(app, engine)
setup_waygate_docs(app, engine)
app.mount("/waygate", WaygateAdmin(engine=engine, auth=("admin", "secret")))
Components¶
WaygateMiddleware¶
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, WaygateRouter, or routes added directly to app).
| Decorator | Import | Behaviour |
|---|---|---|
@maintenance(reason, start, end) |
waygate.fastapi |
503 temporarily |
@disabled(reason) |
waygate.fastapi |
503 permanently |
@env_only(*envs) |
waygate.fastapi |
404 in other envs |
@deprecated(sunset, use_instead) |
waygate.fastapi |
200 + headers |
@force_active |
waygate.fastapi |
Always 200 |
@rate_limit("100/minute") |
waygate.fastapi.decorators |
429 when exceeded |
See Reference: Decorators for full details.
WaygateRouter¶
A drop-in replacement for APIRouter that automatically registers route metadata with the engine at startup.
from waygate.fastapi import WaygateRouter
router = WaygateRouter(engine=engine)
@router.get("/payments")
@maintenance(reason="DB migration")
async def get_payments():
return {"payments": []}
app.include_router(router)
Note
WaygateRouter is optional. WaygateMiddleware also registers routes by scanning app.routes at startup (lazy, on first request). Use WaygateRouter for explicit control over registration order.
WaygateAdmin¶
Mounts the admin dashboard UI and the REST API (used by the waygate CLI) under a single path.
from waygate.fastapi import WaygateAdmin
app.mount("/waygate", WaygateAdmin(engine=engine, auth=("admin", "secret")))
See Tutorial: Admin Dashboard for full details.
Rate limiting¶
Requires waygate[rate-limit] on the server.
from waygate.fastapi 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 waygate 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 (waygate rl and waygate rate-limits are aliases):
See Tutorial: Rate Limiting and Reference: Rate Limiting for full details.
Dependency injection¶
Waygate decorators work as FastAPI Depends() dependencies for per-handler enforcement without middleware.
two patterns
from fastapi import Depends
from waygate.fastapi import disabled, maintenance
# Pattern A — decorator (relies on WaygateMiddleware 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 waygate.fastapi 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 WaygateMiddleware |
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(WaygateMiddleware, engine=engine)
Testing¶
tests/test_payments.py
import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from waygate import MemoryBackend
from waygate import WaygateEngine
from waygate.fastapi import maintenance, force_active
from waygate.fastapi import WaygateMiddleware
from waygate.fastapi import WaygateRouter
async def test_maintenance_returns_503():
engine = WaygateEngine(backend=MemoryBackend())
app = FastAPI()
app.add_middleware(WaygateMiddleware, engine=engine)
router = WaygateRouter(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 waygate 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 = WaygateEngine(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 + WaygateAdmin
Demonstrates every decorator (@maintenance, @disabled, @env_only, @force_active, @deprecated) together with the WaygateAdmin 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/waygate/ (admin / secret)
# Audit log: http://localhost:8000/waygate/audit
CLI quick-start:
waygate login admin # password: secret
waygate status
waygate disable GET:/payments --reason "hotfix"
waygate enable GET:/payments
Full source:
examples/fastapi/basic.py
"""FastAPI — Basic Usage Example.
Demonstrates the core waygate decorators together with the WaygateAdmin
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/waygate/ — admin dashboard (login: admin / secret)
http://localhost:8000/waygate/audit — audit log
CLI quick-start (auto-discovers the server URL):
waygate login admin # password: secret
waygate status
waygate disable /payments --reason "hotfix"
waygate 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 waygate import make_engine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
deprecated,
disabled,
env_only,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = WaygateRouter(engine=engine)
# ---------------------------------------------------------------------------
# Routes with waygate decorators
# ---------------------------------------------------------------------------
@router.get("/health")
@force_active
async def health():
"""Always 200 — bypasses every waygate 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="waygate — Basic Example",
description=(
"Core decorators: `@maintenance`, `@disabled`, `@env_only`, "
"`@force_active`, `@deprecated`.\n\n"
f"Current environment: **{CURRENT_ENV}**"
),
)
app.add_middleware(WaygateMiddleware, engine=engine)
app.include_router(router)
apply_waygate_to_openapi(app, engine)
# Mount the unified admin interface:
# - Dashboard UI → http://localhost:8000/waygate/
# - REST API → http://localhost:8000/waygate/api/... (used by the CLI)
#
# auth= accepts:
# ("user", "pass") — single user
# [("alice","a1"),("bob","b2")] — multiple users
# MyAuthBackend() — custom WaygateAuthBackend 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(
"/waygate",
WaygateAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/waygate",
# secret_key="change-me-in-production",
# token_expiry=86400, # seconds — default 24 h
),
)
Dependency injection¶
Waygate decorators as FastAPI Depends()
Shows how to use waygate decorators as Depends() instead of (or alongside) middleware. Once configure_waygate(app, engine) is called — or WaygateMiddleware 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 waygate 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/waygate/ (admin / secret)
Try it:
curl -i http://localhost:8000/payments # → 503
waygate 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 waygate decorators as FastAPI ``Depends()`` dependencies
instead of (or alongside) the middleware model — but not both on the
same route. Pick one: decorator (with WaygateMiddleware) or Depends()
(without middleware).
Call ``configure_waygate(app, engine)`` once and all decorator deps find the
engine automatically via ``request.app.state.waygate_engine`` — no ``engine=``
argument per route. ``WaygateMiddleware`` calls ``configure_waygate``
automatically at ASGI startup, so if you use middleware you don't need to
call it manually.
Use either the decorator (with WaygateMiddleware) 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; waygate 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/waygate/ — login: admin / secret
CLI quick-start:
waygate login admin # password: secret
waygate status # see all route states
waygate enable /payments # toggle off maintenance without redeploy
waygate disable /payments --reason "emergency patch"
Try these requests:
curl -i http://localhost:8000/payments # → 503 MAINTENANCE_MODE
waygate 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 waygate import make_engine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
deprecated,
disabled,
env_only,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = WaygateRouter(engine=engine)
# ---------------------------------------------------------------------------
# App assembly — configure_waygate is called automatically by WaygateMiddleware
# ---------------------------------------------------------------------------
app = FastAPI(
title="waygate — Dependency Injection Example",
description=(
"``configure_waygate(app, engine)`` called once — no ``engine=`` per route.\n\n"
f"Current environment: **{CURRENT_ENV}**"
),
)
# WaygateMiddleware auto-calls configure_waygate(app, engine) at ASGI startup.
# Without middleware: from waygate.fastapi import configure_waygate
# configure_waygate(app, engine)
app.add_middleware(WaygateMiddleware, 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 WaygateMiddleware.
@router.get(
"/payments",
dependencies=[Depends(maintenance(reason="Scheduled DB migration"))],
)
async def get_payments():
"""503 on startup; toggle off with: waygate 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: waygate 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_waygate_to_openapi(app, engine)
# ---------------------------------------------------------------------------
# Admin interface — dashboard UI + REST API (used by the CLI)
# ---------------------------------------------------------------------------
app.mount(
"/waygate",
WaygateAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/waygate",
# 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 waygate 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 waygate import MaintenanceWindow, MemoryBackend, WaygateEngine
from waygate.fastapi import WaygateMiddleware, WaygateRouter, force_active
engine = WaygateEngine(backend=MemoryBackend())
router = WaygateRouter(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 waygate 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 WaygateEngine automatically
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(_: FastAPI):
await engine.start_scheduler()
yield
await engine.stop_scheduler()
app = FastAPI(
title="waygate — 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(WaygateMiddleware, 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 waygate import MemoryBackend, WaygateEngine
from waygate.fastapi import WaygateMiddleware, WaygateRouter, force_active
engine = WaygateEngine(backend=MemoryBackend())
router = WaygateRouter(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="waygate — 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(WaygateMiddleware, 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/waygate/ (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 ``WaygateMiddleware`` 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/waygate/ — 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 waygate import make_engine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
disabled,
force_active,
maintenance,
)
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = WaygateRouter(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 waygate 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="waygate — 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**: `WaygateMiddleware(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(
WaygateMiddleware,
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_waygate_to_openapi(app, engine)
app.mount(
"/waygate",
WaygateAdmin(engine=engine, auth=("admin", "secret"), prefix="/waygate"),
)
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
WaygateAdmin - Waygate Server mode — build the engine explicitly and register on it before passing to
WaygateAdmin; SDK service apps never fire webhooks
# Waygate Server mode
from waygate import WaygateEngine
from waygate import SlackWebhookFormatter
from waygate.fastapi import WaygateAdmin
engine = WaygateEngine(backend=RedisBackend(...))
engine.add_webhook("https://hooks.slack.com/...", formatter=SlackWebhookFormatter())
waygate_app = WaygateAdmin(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/waygate/ (admin / secret)
Trigger events:
waygate config set-url http://localhost:8000/waygate
waygate login admin # password: secret
waygate disable GET:/payments --reason "hotfix"
waygate enable GET:/payments
waygate maintenance GET:/orders --reason "stock sync"
waygate 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 waygate 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:
waygate config set-url http://localhost:8000/waygate
waygate login admin # password: secret
waygate disable GET:/payments --reason "hotfix"
waygate enable GET:/payments
waygate maintenance GET:/orders --reason "stock sync"
waygate 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 waygate import RouteState, SlackWebhookFormatter, default_formatter, make_engine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
disabled,
force_active,
maintenance,
)
# ---------------------------------------------------------------------------
# Engine & router
# ---------------------------------------------------------------------------
CURRENT_ENV = os.getenv("APP_ENV", "dev")
engine = make_engine(current_env=CURRENT_ENV)
router = WaygateRouter(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": "waygate",
"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 — waygate</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 waygateed 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: waygate 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="waygate — 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(WaygateMiddleware, engine=engine)
app.include_router(router)
apply_waygate_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(
"/waygate",
WaygateAdmin(engine=engine, auth=("admin", "secret"), prefix="/waygate"),
)
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 waygate[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 "waygate[all,rate-limit]"
uv run uvicorn examples.fastapi.rate_limiting:app --reload
# Admin dashboard: http://localhost:8000/waygate/ (admin / secret)
# Rate limits tab: http://localhost:8000/waygate/rate-limits
# Blocked log: http://localhost:8000/waygate/blocked
CLI quick-start:
waygate login admin
waygate rl list
waygate rl set GET:/public/posts 20/minute # raise limit live
waygate rl reset GET:/public/posts # clear counters
waygate 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
* 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/waygate/
CLI — view and mutate policies at runtime (no redeploy needed):
waygate login admin
waygate rl list
waygate rl set GET:/public/posts 20/minute # raise the limit live
waygate rl set POST:/data 10/second --algorithm fixed_window
waygate rl hits # blocked requests log
waygate rl reset GET:/public/posts # clear counters
waygate rl delete GET:/public/posts # remove persisted override
"""
from __future__ import annotations
from fastapi import FastAPI, Request
from waygate import make_engine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
maintenance,
rate_limit,
setup_waygate_docs,
)
engine = make_engine()
router = WaygateRouter(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"}
# ---------------------------------------------------------------------------
# 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:
waygate maintenance /checkout --reason "upgrade" # put in maintenance
# hammer the endpoint — counters stay at 0
waygate 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="waygate — 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(WaygateMiddleware, engine=engine)
app.include_router(router)
apply_waygate_to_openapi(app, engine)
setup_waygate_docs(app, engine)
app.mount(
"/waygate",
WaygateAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/waygate",
),
)
Waygate Server (single service)¶
Centralized Waygate Server + one service via WaygateSDK
Demonstrates the centralized Waygate Server architecture: one Waygate Server process owns all route state, and one service app connects via WaygateSDK. State is enforced locally — zero per-request network overhead.
Two ASGI apps — run each in its own terminal:
# Waygate Server (port 8001)
uv run uvicorn examples.fastapi.waygate_server:waygate_app --port 8001 --reload
# Service app (port 8000)
uv run uvicorn examples.fastapi.waygate_server:service_app --port 8000 --reload
Then visit:
http://localhost:8001/— Waygate 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 = WaygateSDK(
server_url="http://localhost:8001",
app_id="payments-service",
username="admin",
password="secret", # inject from env in production
)
# Option 2 — Pre-issued token
sdk = WaygateSDK(
server_url="http://localhost:8001",
app_id="payments-service",
token="<token-from-waygate-login>",
)
# Option 3 — No auth on the Waygate Server
sdk = WaygateSDK(server_url="http://localhost:8001", app_id="payments-service")
CLI — always targets the Waygate Server:
waygate config set-url http://localhost:8001
waygate login admin # password: secret
waygate status
waygate enable /api/payments
waygate disable /api/orders --reason "hotfix"
waygate maintenance /api/payments --reason "DB migration"
waygate audit
Full source:
Waygate Server (multi-service)¶
Two independent services sharing one Waygate Server
Demonstrates two independent FastAPI services (payments-service and orders-service) both connecting to the same Waygate Server. Each service registers its routes under its own app_id namespace so the dashboard service dropdown and CLI WAYGATE_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 Waygate Server is configured with separate expiry times for human sessions and service tokens:
waygate_app = WaygateServer(
backend=MemoryBackend(),
auth=("admin", "secret"),
token_expiry=3600, # dashboard / CLI: 1 hour
sdk_token_expiry=31536000, # SDK services: 1 year
)
payments_sdk = WaygateSDK(
server_url="http://waygate-server:9000",
app_id="payments-service",
username="admin",
password="secret", # inject from env in production
)
Three ASGI apps — run each in its own terminal:
# Waygate Server (port 8001)
uv run uvicorn examples.fastapi.multi_service:waygate_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/— Waygate 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:
waygate config set-url http://localhost:8001
waygate login admin # password: secret
waygate services # list all connected services
# Scope to payments via env var
export WAYGATE_SERVICE=payments-service
waygate status
waygate enable /api/payments
waygate current-service # confirm active context
# Switch to orders with explicit flag (overrides env var)
waygate status --service orders-service
waygate disable /api/cart --reason "redesign" --service orders-service
# Unscoped — operates across all services
unset WAYGATE_SERVICE
waygate status
waygate audit
waygate global disable --reason "emergency maintenance"
waygate global enable
Full source:
examples/fastapi/multi_service.py
"""FastAPI — Multi-Service Waygate Server Example.
Demonstrates two independent FastAPI services (payments and orders) both
connecting to the same Waygate 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:
Waygate Server (port 8001):
uv run --with uvicorn uvicorn examples.fastapi.multi_service:waygate_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/ — Waygate 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 Waygate Server; use --service or WAYGATE_SERVICE to scope:
# One-time setup
waygate config set-url http://localhost:8001
waygate login admin # password: secret
# View all registered services
waygate services
# Manage payments routes
export WAYGATE_SERVICE=payments-service
waygate status
waygate disable /api/payments --reason "hotfix"
waygate enable /api/payments
# Switch to orders without changing env var
waygate status --service orders-service
# Explicit --service flag overrides the env var
export WAYGATE_SERVICE=payments-service
waygate enable /api/orders --service orders-service
# Clear the env var to work across all services at once
unset WAYGATE_SERVICE
waygate status # shows routes from both services
waygate audit # audit log from both services
# Global maintenance — affects ALL services
waygate global disable --reason "emergency maintenance"
waygate 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 Waygate Server only. All SDK clients receive
live SSE updates regardless of backend — they connect to the Waygate Server
over HTTP, never to the backend directly:
* MemoryBackend — fine for development; state lost on Waygate Server restart.
* FileBackend — state survives restarts; single Waygate Server instance only.
* RedisBackend — needed only when running multiple Waygate 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 Waygate Server restarts.
* Prefer passing username/password to each WaygateSDK 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 waygate import MemoryBackend
from waygate.fastapi import (
WaygateRouter,
apply_waygate_to_openapi,
deprecated,
disabled,
force_active,
maintenance,
setup_waygate_docs,
)
from waygate.sdk import WaygateSDK
from waygate.server import WaygateServer
# ---------------------------------------------------------------------------
# Waygate Server — shared by all services
# ---------------------------------------------------------------------------
# Run: uv run uvicorn examples.fastapi.multi_service:waygate_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 waygate import RedisBackend
# backend = RedisBackend("redis://localhost:6379")
waygate_app = WaygateServer(
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
# Waygate Server. The dashboard shows them separately from orders-service.
# CLI: export WAYGATE_SERVICE=payments-service; waygate status
payments_sdk = WaygateSDK(
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["WAYGATE_USERNAME"]
# password="secret", # inject from env: os.environ["WAYGATE_PASSWORD"]
# Or use a pre-issued token: token=os.environ["WAYGATE_TOKEN"]
reconnect_delay=5.0,
)
payments_app = FastAPI(
title="waygate — Payments Service",
description=(
"Connects to the Waygate Server at **http://localhost:8001** as "
"`payments-service`. Manage routes from the "
"[Waygate Dashboard](http://localhost:8001/) or via the CLI with "
"`export WAYGATE_SERVICE=payments-service`."
),
)
payments_sdk.attach(payments_app)
payments_router = WaygateRouter(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 WAYGATE_SERVICE=payments-service
waygate 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:
waygate 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_waygate_to_openapi(payments_app, payments_sdk.engine)
setup_waygate_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 WAYGATE_SERVICE=orders-service; waygate status
orders_sdk = WaygateSDK(
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["WAYGATE_USERNAME"]
# password="secret", # inject from env: os.environ["WAYGATE_PASSWORD"]
# Or use a pre-issued token: token=os.environ["WAYGATE_TOKEN"]
reconnect_delay=5.0,
)
orders_app = FastAPI(
title="waygate — Orders Service",
description=(
"Connects to the Waygate Server at **http://localhost:8001** as "
"`orders-service`. Manage routes from the "
"[Waygate Dashboard](http://localhost:8001/) or via the CLI with "
"`export WAYGATE_SERVICE=orders-service`."
),
)
orders_sdk.attach(orders_app)
orders_router = WaygateRouter(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:
waygate 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:
waygate 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:
waygate maintenance /api/cart --reason "cart redesign" \\
--service orders-service
"""
return {"cart": {"items": [], "total": 0.0}}
orders_app.include_router(orders_router)
apply_waygate_to_openapi(orders_app, orders_sdk.engine)
setup_waygate_docs(orders_app, orders_sdk.engine)
# ---------------------------------------------------------------------------
# CLI reference — multi-service workflow
# ---------------------------------------------------------------------------
#
# Setup (once):
# waygate config set-url http://localhost:8001
# waygate login admin
#
# View all services and their routes:
# waygate services
# waygate status # routes from ALL services combined
#
# Scope to a specific service via env var:
# export WAYGATE_SERVICE=payments-service
# waygate status # only payments-service routes
# waygate disable /api/payments --reason "hotfix"
# waygate enable /api/payments
# waygate maintenance /api/refunds --reason "audit"
#
# Switch service without changing the env var (--service flag):
# waygate status --service orders-service
# waygate disable /api/orders --reason "inventory sync" \\
# --service orders-service
#
# Explicit flag always overrides the WAYGATE_SERVICE env var:
# export WAYGATE_SERVICE=payments-service
# waygate enable /api/orders --service orders-service # acts on orders
#
# Unscoped commands operate across all services:
# unset WAYGATE_SERVICE
# waygate audit # audit log from both services
# waygate global disable --reason "emergency maintenance"
# waygate global enable
Custom backend (SQLite)¶
Implementing WaygateBackend 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/waygate/ (admin / secret)
# Audit log: http://localhost:8000/waygate/audit
CLI quick-start:
waygate config set-url http://localhost:8000/waygate
waygate login admin # password: secret
waygate status
waygate disable GET:/payments --reason "hotfix"
waygate enable GET:/payments
waygate log
Full source:
examples/fastapi/custom_backend/sqlite_backend.py
"""Custom Backend Example — SQLite via aiosqlite.
This file shows how to wire waygate to a storage layer it does not ship
with by implementing the ``WaygateBackend`` abstract base class.
The contract is simple: implement six async methods and waygate 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/waygate/ — admin dashboard (login: admin / secret)
http://localhost:8000/waygate/audit — audit log
CLI quick-start (the CLI talks to the app's admin API — it never touches
the database directly):
waygate config set-url http://localhost:8000/waygate
waygate login admin # password: secret
waygate status
waygate disable GET:/payments --reason "hotfix"
waygate enable GET:/payments
waygate log
"""
from __future__ import annotations
import sqlite3
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
import aiosqlite
from fastapi import FastAPI
from waygate import AuditEntry, RouteState, WaygateBackend, WaygateEngine
from waygate.fastapi import (
WaygateAdmin,
WaygateMiddleware,
WaygateRouter,
apply_waygate_to_openapi,
disabled,
force_active,
maintenance,
)
# ---------------------------------------------------------------------------
# SQLiteBackend — implements the WaygateBackend contract
#
# Rules to follow when building any custom backend:
#
# 1. Subclass ``WaygateBackend`` from ``waygate.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 — WaygateEngine 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 waygate_states (
path TEXT PRIMARY KEY,
state_json TEXT NOT NULL
)
"""
_CREATE_AUDIT_TABLE = """
CREATE TABLE IF NOT EXISTS waygate_audit (
id TEXT PRIMARY KEY,
timestamp TEXT NOT NULL,
path TEXT NOT NULL,
entry_json TEXT NOT NULL
)
"""
class SQLiteBackend(WaygateBackend):
"""waygate 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("waygate-state.db")
>>> engine = WaygateEngine(backend=backend)
>>> async with engine: # calls startup() then shutdown()
... states = await engine.list_states()
"""
def __init__(self, db_path: str = "waygate-state.db") -> None:
self._db_path = db_path
self._db: aiosqlite.Connection | None = None
# ------------------------------------------------------------------
# Lifecycle hooks — called automatically by WaygateEngine
# ------------------------------------------------------------------
async def startup(self) -> None:
"""Open the database connection and create tables if needed.
Called automatically by ``WaygateEngine.__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 ``WaygateEngine.__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
# ------------------------------------------------------------------
# WaygateBackend — 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 waygate_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 waygate_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 waygate_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 waygate_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 waygate_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 waygate_audit
WHERE id NOT IN (
SELECT id FROM waygate_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 waygate_audit
WHERE path = ?
ORDER BY timestamp DESC
LIMIT ?
"""
params: tuple[object, ...] = (path, limit)
else:
query = """
SELECT entry_json FROM waygate_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("waygate-state.db")
engine = WaygateEngine(backend=backend)
router = WaygateRouter(engine=engine)
@router.get("/health")
@force_active
async def health():
"""Always 200 — bypasses every waygate 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="waygate — SQLite Custom Backend Example",
description=(
"All route state and audit log entries are persisted in `waygate-state.db`. "
"Restart the server and the state survives.\n\n"
"Admin UI and CLI API available at `/waygate/`."
),
lifespan=lifespan,
)
app.add_middleware(WaygateMiddleware, engine=engine)
app.include_router(router)
apply_waygate_to_openapi(app, engine)
# Mount the unified admin interface:
# - Dashboard UI → http://localhost:8000/waygate/
# - REST API → http://localhost:8000/waygate/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(
"/waygate",
WaygateAdmin(
engine=engine,
auth=("admin", "secret"),
prefix="/waygate",
# secret_key="change-me-in-production",
# token_expiry=86400, # seconds — default 24 h
),
)