Route lifecycle management for Python web frameworks — maintenance mode, environment gating, deprecation, canary rollouts, and more. No restarts required.
Most "maintenance mode" tools are blunt instruments: shut everything down or nothing at all. api-shield treats each route as a first-class entity with its own lifecycle. State changes take effect immediately through middleware — no redeployment, no server restart.
uv add api-shield
# or: pip install api-shieldFor the full feature set:
uv add "api-shield[all]"from fastapi import FastAPI
from shield.core.config import make_engine
from shield.fastapi import (
ShieldMiddleware,
apply_shield_to_openapi,
setup_shield_docs,
maintenance,
env_only,
disabled,
force_active,
deprecated,
)
engine = make_engine() # reads SHIELD_BACKEND, SHIELD_ENV, etc.
app = FastAPI(title="My API")
app.add_middleware(ShieldMiddleware, engine=engine)
@app.get("/payments")
@maintenance(reason="Database migration — back at 04:00 UTC")
async def get_payments():
return {"payments": []}
@app.get("/health")
@force_active # always 200, immune to all shield checks
async def health():
return {"status": "ok"}
@app.get("/debug")
@env_only("dev", "staging") # silent 404 in production
async def debug():
return {"debug": True}
@app.get("/old-endpoint")
@disabled(reason="Use /v2/endpoint")
async def old_endpoint():
return {}
@app.get("/v1/users")
@deprecated(sunset="Sat, 01 Jan 2027 00:00:00 GMT", use_instead="/v2/users")
async def v1_users():
return {"users": []}
apply_shield_to_openapi(app, engine) # filter /docs and /redoc
setup_shield_docs(app, engine) # inject maintenance banners into UIGET /payments → 503 {"error": {"code": "MAINTENANCE_MODE", "reason": "..."}}
GET /health → 200 always (force_active)
GET /debug → 404 in production (env_only)
GET /old-endpoint → 503 {"error": {"code": "ROUTE_DISABLED", "reason": "..."}}
GET /v1/users → 200 + Deprecation/Sunset/Link response headers
Incoming HTTP request
│
▼
ShieldMiddleware.dispatch()
│
├─ /docs, /redoc, /openapi.json ──────────────────────→ pass through
│
├─ Lazy-scan app routes for __shield_meta__ (once only)
│
├─ @force_active route? ──────────────────────────────→ pass through
│ (unless global maintenance overrides — see below)
│
├─ engine.check(path, method)
│ │
│ ├─ Global maintenance ON + path not exempt? → 503
│ ├─ MAINTENANCE → 503 + Retry-After header
│ ├─ DISABLED → 503
│ ├─ ENV_GATED → 404 (silent — path existence not revealed)
│ ├─ DEPRECATED → pass through + inject response headers
│ └─ ACTIVE → pass through ✓
│
└─ call_next(request)
Shield decorators stamp __shield_meta__ on the endpoint function. This metadata is registered with the engine at startup via two mechanisms:
- ASGI lifespan interception —
ShieldMiddlewarehooks intolifespan.startup.completeto scan all app routes before the first request. This works with anyAPIRouter(plain orShieldRouter). - Lazy fallback — on the first HTTP request if no lifespan was triggered (e.g. test environments).
State registration is persistence-first: if the backend already has a state for a route (written by a previous CLI command or earlier server run), the decorator default is ignored and the persisted state wins. This means runtime changes survive restarts.
All decorators work on any router type — plain APIRouter, ShieldRouter, or routes added directly to the FastAPI app instance.
Puts a route into maintenance mode. Returns 503 with a structured JSON body. If start/end are provided, the maintenance window is also stored for scheduling.
from shield.fastapi import maintenance
from datetime import datetime, UTC
@router.get("/payments")
@maintenance(reason="DB migration in progress")
async def get_payments():
...
# With a scheduled window
@router.post("/orders")
@maintenance(
reason="Order system upgrade",
start=datetime(2025, 6, 1, 2, 0, tzinfo=UTC),
end=datetime(2025, 6, 1, 4, 0, tzinfo=UTC),
)
async def create_order():
...Response:
{
"error": {
"code": "MAINTENANCE_MODE",
"message": "This endpoint is temporarily unavailable",
"reason": "DB migration in progress",
"path": "/payments",
"retry_after": "2025-06-01T04:00:00Z"
}
}Permanently disables a route. Returns 503. Use for routes that should never be called again (migrations, removed features).
from shield.fastapi import disabled
@router.get("/legacy/report")
@disabled(reason="Replaced by /v2/reports — update your clients")
async def legacy_report():
...Restricts a route to specific environment names. In any other environment the route returns a silent 404 — it does not reveal that the path exists.
from shield.fastapi import env_only
@router.get("/internal/metrics")
@env_only("dev", "staging")
async def internal_metrics():
...The current environment is set via SHIELD_ENV or when constructing the engine:
engine = ShieldEngine(current_env="production")
# or
engine = make_engine(current_env="staging")Bypasses all shield checks. Use for health checks, status endpoints, and any route that must always be reachable.
from shield.fastapi import force_active
@router.get("/health")
@force_active
async def health():
return {"status": "ok"}@force_active routes are also immune to runtime changes — you cannot disable or put them in maintenance via the CLI or engine. This is intentional: health check routes must be trustworthy.
The only exception is when global maintenance mode is enabled with include_force_active=True (see Global Maintenance Mode).
Marks a route as deprecated. Requests still succeed, but the middleware injects RFC-compliant response headers:
from shield.fastapi import deprecated
@router.get("/v1/users")
@deprecated(
sunset="Sat, 01 Jan 2027 00:00:00 GMT",
use_instead="/v2/users",
)
async def v1_users():
return {"users": []}Response headers added automatically:
Deprecation: true
Sunset: Sat, 01 Jan 2027 00:00:00 GMT
Link: </v2/users>; rel="successor-version"
The route is also marked deprecated: true in the OpenAPI schema and shown with a visual indicator in /docs.
Global maintenance blocks every route with a single call, without requiring per-route decorators. Use it for full deployments, infrastructure work, or emergency stops.
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Enable global maintenance at startup
await engine.enable_global_maintenance(
reason="Scheduled deployment — back in 15 minutes",
exempt_paths=["/health", "GET:/admin/status"],
include_force_active=False, # @force_active routes still bypass (default)
)
yield
await engine.disable_global_maintenance()Or toggle at runtime via any async context:
# Enable — all non-exempt routes return 503 immediately
await engine.enable_global_maintenance(reason="Emergency patch")
# Disable — routes return to their per-route state
await engine.disable_global_maintenance()
# Check current state
cfg = await engine.get_global_maintenance()
print(cfg.enabled, cfg.reason, cfg.exempt_paths)
# Add/remove individual exemptions without toggling the mode
await engine.set_global_exempt_paths(["/health", "/status"])# Enable with exemptions
shield global enable \
--reason "Scheduled deployment" \
--exempt /health \
--exempt GET:/admin/status
# Block even force_active routes
shield global enable --reason "Hard lockdown" --include-force-active
# Add/remove exemptions while maintenance is already active
shield global exempt-add /monitoring/ping
shield global exempt-remove /monitoring/ping
# Check current state
shield global status
# Disable
shield global disable| Option | Default | Description |
|---|---|---|
reason |
"" |
Shown in every 503 response body |
exempt_paths |
[] |
Bare paths (/health) or method-prefixed (GET:/health) |
include_force_active |
False |
When True, @force_active routes are also blocked |
from shield.fastapi import apply_shield_to_openapi
apply_shield_to_openapi(app, engine)Effect on /docs and /redoc:
| Route status | Schema behaviour |
|---|---|
DISABLED |
Hidden from all schemas |
ENV_GATED (wrong env) |
Hidden from all schemas |
MAINTENANCE |
Visible; operation summary prefixed with 🔧; description shows warning block; x-shield-status extension added |
DEPRECATED |
Marked deprecated: true; successor path shown |
ACTIVE |
No change |
Schema is computed fresh on every request — runtime state changes (CLI, engine calls) reflect immediately without restarting.
from shield.fastapi import setup_shield_docs
apply_shield_to_openapi(app, engine) # must come first
setup_shield_docs(app, engine)Replaces both /docs and /redoc with enhanced versions:
Global maintenance ON:
- Full-width pulsing red sticky banner at the top of the page
- Reason text and exempt paths displayed
- Refreshes automatically every 15 seconds — no page reload needed
Global maintenance OFF:
- Small green "All systems operational" chip in the bottom-right corner
Per-route maintenance:
- Orange left-border on the operation block
🔧 MAINTENANCEbadge appended to the summary bar
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"pyproject.toml includes:
[tool.pytest.ini_options]
asyncio_mode = "auto" # all async tests work without @pytest.mark.asyncioRun tests:
uv run pytest # all tests
uv run pytest -v # verbose
uv run pytest tests/fastapi/test_middleware.py # specific file
uv run pytest tests/core/ # core only (no FastAPI dependency)Runnable examples are in examples/fastapi/.
| File | What it demonstrates |
|---|---|
| basic.py | Core decorators: @maintenance, @disabled, @env_only, @force_active, @deprecated |
| scheduled_maintenance.py | Auto-activating maintenance windows via schedule_maintenance() |
| global_maintenance.py | Blocking every route at once with enable_global_maintenance() |
| custom_backend/sqlite_backend.py | Full custom backend implementation using SQLite |
Run any example:
# Basic decorators
uv run uvicorn examples.fastapi.basic:app --reload
# Scheduled maintenance window
uv run uvicorn examples.fastapi.scheduled_maintenance:app --reload
# Global maintenance mode
uv run uvicorn examples.fastapi.global_maintenance:app --reload
# SQLite custom backend (requires: pip install aiosqlite)
uv run uvicorn examples.fastapi.custom_backend.sqlite_backend:app --reloadDjango adapter is planned. It will provide:
ShieldMiddlewareas a standard Django middleware class- Same decorators (
@maintenance,@disabled,@env_only,@deprecated,@force_active) usable on Django views and DRF viewsets - Integration with Django's URL routing for route registration at startup
- DRF schema filtering for
drf-spectacular/drf-yasg
Track progress: github.com/Attakay78/api-shield
Flask adapter is planned. It will provide:
ShieldMiddlewareas a WSGI/ASGI middleware compatible with Flask- Same decorators usable on Flask route functions and Blueprints
- Integration with Flask's URL map for route registration at startup
- OpenAPI schema filtering for
flask-openapi3/flasgger
Track progress: github.com/Attakay78/api-shield
The backend determines where route state and the audit log are persisted. Backends are shared across all adapters.
In-process dict. No persistence across restarts. CLI cannot share state with the running server.
from shield.core.backends.memory import MemoryBackend
engine = ShieldEngine(backend=MemoryBackend())Best for: development, single-process testing.
JSON file on disk. Survives restarts. CLI shares state with the running server when both point to the same file.
from shield.core.backends.file import FileBackend
engine = ShieldEngine(backend=FileBackend(path="shield-state.json"))Or via environment variable:
SHIELD_BACKEND=file SHIELD_FILE_PATH=./shield-state.json uvicorn app:appBest for: single-instance deployments, simple setups, CLI-driven workflows.
Redis via redis-py async. Supports multi-instance deployments. CLI changes reflect immediately on all running instances.
from shield.core.backends.redis import RedisBackend
engine = ShieldEngine(backend=RedisBackend(url="redis://localhost:6379/0"))Or via environment variable:
SHIELD_BACKEND=redis SHIELD_REDIS_URL=redis://localhost:6379/0 uvicorn app:appKey schema:
shield:state:{path}— route stateshield:audit— audit log (LPUSH, capped at 1000 entries)shield:global— global maintenance configuration
Best for: multi-instance / load-balanced deployments, production.
Any storage layer can be used as a backend by subclassing ShieldBackend and implementing six async methods. api-shield handles everything else — the engine, middleware, decorators, CLI, and audit log all work unchanged.
from shield.core.backends.base import ShieldBackend
from shield.core.models import AuditEntry, RouteState
class MyBackend(ShieldBackend):
async def get_state(self, path: str) -> RouteState:
# Return stored state. MUST raise KeyError if path not found.
...
async def set_state(self, path: str, state: RouteState) -> None:
# Persist state for path, overwriting any existing entry.
...
async def delete_state(self, path: str) -> None:
# Remove state for path. No-op if not found.
...
async def list_states(self) -> list[RouteState]:
# Return all registered route states.
...
async def write_audit(self, entry: AuditEntry) -> None:
# Append entry to the audit log.
...
async def get_audit_log(
self, path: str | None = None, limit: int = 100
) -> list[AuditEntry]:
# Return audit entries newest-first, optionally filtered by path.
...subscribe() is optional. The base class raises NotImplementedError by default, and the dashboard falls back to polling. Override it if your backend supports pub/sub.
RouteState and AuditEntry are Pydantic v2 models. Use their built-in helpers:
# Serialise to a JSON string for storage
json_str = state.model_dump_json()
# Deserialise from a JSON string
state = RouteState.model_validate_json(json_str)| Rule | Detail |
|---|---|
get_state() must raise KeyError |
The engine uses KeyError to distinguish "not registered" from "registered but active" |
| Fail-open on errors | Let exceptions bubble up — ShieldEngine wraps every backend call and allows requests through on failure |
| Thread safety | All methods are async; use your storage library's async client where available |
| Global maintenance | Inherited from ShieldBackend base class — no extra work needed unless you want a dedicated storage path |
| Lifecycle hooks | Override startup() / shutdown() for async setup/teardown — called automatically by async with engine: |
from shield.core.engine import ShieldEngine
backend = MyBackend()
engine = ShieldEngine(backend=backend)Use async with engine: to call startup() and shutdown() automatically:
# FastAPI lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
async with engine: # → backend.startup() … backend.shutdown()
yield
app = FastAPI(lifespan=lifespan)From there everything works as normal — decorators, middleware, CLI, audit log.
Set SHIELD_BACKEND=custom and point SHIELD_CUSTOM_PATH at a zero-arg factory function that returns your configured backend instance:
# myapp/backends.py
import os
from myapp.db import MyBackend
def make_shield_backend() -> MyBackend:
"""Zero-arg factory — reads config from environment variables."""
return MyBackend(dsn=os.environ["MY_DB_DSN"])# One-off
SHIELD_BACKEND=custom \
SHIELD_CUSTOM_PATH=myapp.backends:make_shield_backend \
MY_DB_DSN=postgresql://localhost/myapp \
shield statusOr configure it once in your .shield file so you don't repeat it on every command:
# .shield
SHIELD_BACKEND=custom
SHIELD_CUSTOM_PATH=myapp.backends:make_shield_backend
MY_DB_DSN=postgresql://localhost/myappshield status
shield disable GET:/payments --reason "patch"
shield logThe CLI calls startup() and shutdown() automatically around every command, so your backend's connection lifecycle is handled correctly with no extra work.
A complete working implementation backed by SQLite is in examples/fastapi/custom_backend/sqlite_backend.py.
Key points from that implementation:
import aiosqlite
from shield.core.backends.base import ShieldBackend
from shield.core.models import AuditEntry, RouteState
class SQLiteBackend(ShieldBackend):
def __init__(self, db_path: str = "shield-state.db") -> None:
self._db_path = db_path
self._db: aiosqlite.Connection | None = None
async def connect(self) -> None:
"""Open connection and create tables. Call at app startup."""
self._db = await aiosqlite.connect(self._db_path)
await self._db.execute("""
CREATE TABLE IF NOT EXISTS shield_states (
path TEXT PRIMARY KEY, state_json TEXT NOT NULL
)
""")
await self._db.execute("""
CREATE TABLE IF NOT EXISTS shield_audit (
id TEXT PRIMARY KEY, timestamp TEXT NOT NULL,
path TEXT NOT NULL, entry_json TEXT NOT NULL
)
""")
await self._db.commit()
async def get_state(self, path: str) -> RouteState:
async with self._db.execute(
"SELECT state_json FROM shield_states WHERE path = ?", (path,)
) as cur:
row = await cur.fetchone()
if row is None:
raise KeyError(path) # ← required contract
return RouteState.model_validate_json(row[0])
async def set_state(self, path: str, state: RouteState) -> None:
await self._db.execute(
"INSERT INTO shield_states VALUES (?, ?)"
" ON CONFLICT(path) DO UPDATE SET state_json = excluded.state_json",
(path, state.model_dump_json()),
)
await self._db.commit()
# ... delete_state, list_states, write_audit, get_audit_log
# See the full file for the complete implementation.Run the demo app:
pip install aiosqlite
uv run uvicorn examples.fastapi.custom_backend.sqlite_backend:app --reloadUse it with the CLI:
# One-off
SHIELD_BACKEND=custom \
SHIELD_CUSTOM_PATH=examples.fastapi.custom_backend.sqlite_backend:make_backend \
SHIELD_SQLITE_PATH=shield-state.db \
shield status
# Or in .shield
# SHIELD_BACKEND=custom
# SHIELD_CUSTOM_PATH=examples.fastapi.custom_backend.sqlite_backend:make_backend
# SHIELD_SQLITE_PATH=shield-state.dbThe shield CLI operates on the same backend as the running server. Requires SHIELD_BACKEND=file or SHIELD_BACKEND=redis to share state (the default memory backend is process-local).
uv pip install -e ".[cli]"shield status # all registered routes
shield status GET:/payments # inspect one route
shield enable GET:/payments
shield disable GET:/payments --reason "Security patch"
shield maintenance GET:/payments --reason "DB swap"
shield maintenance GET:/payments \
--reason "DB migration" \
--start 2025-06-01T02:00Z \
--end 2025-06-01T04:00Z
shield schedule GET:/payments \
--start 2025-06-01T02:00Z \
--end 2025-06-01T04:00Z \
--reason "Planned migration"shield global status
shield global enable --reason "Deploying v2" --exempt /health
shield global disable
shield global exempt-add /monitoring
shield global exempt-remove /monitoringshield log # last 20 entries across all routes
shield log --route GET:/payments # filter by route
shield log --limit 100Routes are stored with method-prefixed keys:
| What you type | What gets stored |
|---|---|
@router.get("/payments") |
GET:/payments |
@router.post("/payments") |
POST:/payments |
@router.get("/api/v1/users") |
GET:/api/v1/users |
shield disable "GET:/payments"
shield enable "/payments" # applies to all methodsEvery state change writes an immutable audit entry:
entries = await engine.get_audit_log(limit=50)
entries = await engine.get_audit_log(path="GET:/payments", limit=20)
for e in entries:
print(e.timestamp, e.actor, e.action, e.path,
e.previous_status, "→", e.new_status, e.reason)Fields: id, timestamp, path, action, actor, reason, previous_status, new_status.
The CLI uses getpass.getuser() (the logged-in OS username) as the default actor:
shield disable GET:/payments --reason "Security patch"
# audit entry: actor="alice", action="disable", path="GET:/payments"Both the app and CLI auto-discover a .shield file by walking up from the current directory:
# .shield
SHIELD_BACKEND=file
SHIELD_FILE_PATH=shield-state.json
SHIELD_ENV=productionPriority order (highest wins):
- Explicit constructor arguments
os.environ.shieldfile- Built-in defaults
shield --config /etc/myapp/.shield statusshield/
├── core/ # Framework-agnostic — zero framework imports
│ ├── models.py # RouteState, AuditEntry, GlobalMaintenanceConfig
│ ├── engine.py # ShieldEngine — all business logic
│ ├── scheduler.py # MaintenanceScheduler (asyncio.Task based)
│ ├── config.py # Backend/engine factory + .shield file loading
│ ├── exceptions.py # MaintenanceException, EnvGatedException, ...
│ └── backends/
│ ├── base.py # ShieldBackend ABC
│ ├── memory.py # In-process dict
│ ├── file.py # JSON file via aiofiles
│ └── redis.py # Redis via redis-py async
│
├── fastapi/ # FastAPI adapter
│ ├── middleware.py # ShieldMiddleware (ASGI, BaseHTTPMiddleware)
│ ├── decorators.py # @maintenance, @disabled, @env_only, ...
│ ├── router.py # ShieldRouter + scan_routes()
│ └── openapi.py # Schema filter + docs UI customisation
│
├── adapters/ # Future framework adapters
│ ├── django/ # Coming soon
│ └── flask/ # Coming soon
│
└── cli/
└── main.py # Typer CLI app
shield.corenever imports from any adapter — the core is framework-agnostic and powers all current and future adapters.- All business logic lives in
ShieldEngine— middleware and decorators are transport layers that callengine.check(), never make policy decisions themselves. engine.check()is the single chokepoint — every request, regardless of framework or router type, goes through this one method.- Fail-open on backend errors — if the backend is unreachable, requests pass through. Shield never takes down an API due to its own failures.
- Persistence-first registration — if a route already has persisted state, the decorator default is ignored. Runtime changes survive restarts.
All shield-generated error responses follow a consistent JSON structure:
{
"error": {
"code": "MAINTENANCE_MODE",
"message": "This endpoint is temporarily unavailable",
"reason": "Database migration in progress",
"path": "/api/payments",
"retry_after": "2025-06-01T04:00:00Z"
}
}| Scenario | HTTP status | code |
|---|---|---|
| Route in maintenance | 503 | MAINTENANCE_MODE |
| Route disabled | 503 | ROUTE_DISABLED |
| Route env-gated (wrong env) | 404 | (no body — silent) |
| Global maintenance active | 503 | MAINTENANCE_MODE |