Skip to content

Attakay78/api-shield

Repository files navigation

api-shield

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.


Contents


Adapters

FastAPI

Installation

uv add api-shield
# or: pip install api-shield

For the full feature set:

uv add "api-shield[all]"

Quick Start

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 UI
GET /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

How It Works

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)
Route Registration

Shield decorators stamp __shield_meta__ on the endpoint function. This metadata is registered with the engine at startup via two mechanisms:

  1. ASGI lifespan interceptionShieldMiddleware hooks into lifespan.startup.complete to scan all app routes before the first request. This works with any APIRouter (plain or ShieldRouter).
  2. 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.


Decorators

All decorators work on any router type — plain APIRouter, ShieldRouter, or routes added directly to the FastAPI app instance.

@maintenance(reason, start, end)

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"
  }
}

@disabled(reason)

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():
    ...

@env_only(*envs)

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")

@force_active

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).


@deprecated(sunset, use_instead)

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 Mode

Global maintenance blocks every route with a single call, without requiring per-route decorators. Use it for full deployments, infrastructure work, or emergency stops.

Programmatic (lifespan or runtime)
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"])
Via CLI
# 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
Options
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

OpenAPI & Docs Integration

Schema filtering
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.


Docs UI customisation
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
  • 🔧 MAINTENANCE badge appended to the summary bar

Testing

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.asyncio

Run 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)

Examples

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 --reload

Django — Coming Soon

Django adapter is planned. It will provide:

  • ShieldMiddleware as 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 — Coming Soon

Flask adapter is planned. It will provide:

  • ShieldMiddleware as 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


Backends

The backend determines where route state and the audit log are persisted. Backends are shared across all adapters.

MemoryBackend (default)

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.


FileBackend

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:app

Best for: single-instance deployments, simple setups, CLI-driven workflows.


RedisBackend

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:app

Key schema:

  • shield:state:{path} — route state
  • shield:audit — audit log (LPUSH, capped at 1000 entries)
  • shield:global — global maintenance configuration

Best for: multi-instance / load-balanced deployments, production.


Custom Backends

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.

Contract

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.

Serialisation

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)

Rules

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:

Wire the custom backend to the 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.

CLI access for custom backends

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 status

Or 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/myapp
shield status
shield disable GET:/payments --reason "patch"
shield log

The CLI calls startup() and shutdown() automatically around every command, so your backend's connection lifecycle is handled correctly with no extra work.

SQLite example

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 --reload

Use 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.db

CLI Reference

The 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]"

Route commands

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"

Global maintenance commands

shield global status
shield global enable --reason "Deploying v2" --exempt /health
shield global disable
shield global exempt-add /monitoring
shield global exempt-remove /monitoring

Audit log

shield log                          # last 20 entries across all routes
shield log --route GET:/payments    # filter by route
shield log --limit 100

Route key format

Routes 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 methods

Audit Log

Every 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"

Configuration File

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=production

Priority order (highest wins):

  1. Explicit constructor arguments
  2. os.environ
  3. .shield file
  4. Built-in defaults
shield --config /etc/myapp/.shield status

Architecture

shield/
├── 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

Key design rules

  1. shield.core never imports from any adapter — the core is framework-agnostic and powers all current and future adapters.
  2. All business logic lives in ShieldEngine — middleware and decorators are transport layers that call engine.check(), never make policy decisions themselves.
  3. engine.check() is the single chokepoint — every request, regardless of framework or router type, goes through this one method.
  4. Fail-open on backend errors — if the backend is unreachable, requests pass through. Shield never takes down an API due to its own failures.
  5. Persistence-first registration — if a route already has persisted state, the decorator default is ignored. Runtime changes survive restarts.

Error Response Format

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

About

Route lifecycle management for Python web frameworks — maintenance mode, env gating, deprecation, and more

Resources

License

Contributing

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors