Created
October 4, 2020 22:49
-
-
Save geospatial-jeff/17d677202f1223eacd3b32960ac29c60 to your computer and use it in GitHub Desktop.
FastAPI middleware scoped to a router
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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