Skip to content

Instantly share code, notes, and snippets.

@geospatial-jeff
Created October 4, 2020 22:49
Show Gist options
  • Save geospatial-jeff/17d677202f1223eacd3b32960ac29c60 to your computer and use it in GitHub Desktop.
Save geospatial-jeff/17d677202f1223eacd3b32960ac29c60 to your computer and use it in GitHub Desktop.
FastAPI middleware scoped to a router
from typing import Callable
from fastapi import FastAPI, APIRouter
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.routing import Match
app = FastAPI()
router = APIRouter()
another_router = APIRouter()
@router.get("/hello")
async def hello_world():
return "world"
@another_router.get("/foo")
async def foo_bar():
return "bar"
app.include_router(router)
app.include_router(another_router)
def router_middleware(app: FastAPI, router: APIRouter):
"""Decorator to add a router-specific middleware."""
def deco(func: Callable) -> Callable:
async def _middleware(request: Request, call_next):
# Check if scopes match
matches = any([route.matches(request.scope)[0] == Match.FULL for route in router.routes])
if matches: # Run the middleware if they do
return await func(request, call_next)
else: # Otherwise skip the middleware
return await call_next(request)
app.add_middleware(BaseHTTPMiddleware, dispatch=_middleware)
return func
return deco
@router_middleware(app, router)
async def test(request: Request, call_next):
"""This middleware will execute only for routes attached to the specified router"""
response = await call_next(request)
print("inside middleware")
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment