FastAPI: Experiment Middleware feature
briefly experimenting FastAPI middleware feature
Intro
While I worked on adding authentication into FastAPI application, I had a chance to take a look the FastAPI Middleware feature.
This post is the documentation that I wrote during the experimentation on FastAPI Middleware.
What is Middleware in FastAPI?
Let me repeat what the official FastAPI described about the Middleware.
A “middleware” is a function that works with every request before it is processed by any specific path operation. And also with every response before returning it.
Experiment 1: Build a simple middleware
Let’s try the example in FastAPI documentation. The example is adding API process time into Response Header.
Before adding the middleware that adds the process time into Response Header, let’s try a simple one like below.
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.get("/")
async def root():
return "Wonderful!!"# by using curl, the http response header can be checked.
$ curl -i 127.0.0.1:8000
HTTP/1.1 200 OK
date: Sat, 12 Aug 2023 04:19:04 GMT
server: uvicorn
content-length: 13
content-type: application/json
"Wonderful!!"% Let’s add the custom middleware that adds the “X-Process-Time” into Response Header.
To create a middleware, @app.middleware("http") on top of a function, add_process_time_header, is added.
import time
from fastapi import FastAPI, Request
app = FastAPI()
# Implemented and added custom middleware to FastAPI
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - start_time)
return response
@app.get("/")
async def root():
return "Wonderful!!"I can see the “x-process-time” in the response header like the screenshot below.
Experiment 2: What if there is no matched path? Will the middleware be executed?
I got curious if the middleware is executed only if there is a matched request path.
I added a dummy message that will be printed whenever middleware is executed.
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
print("in add_process_time_header middleware.") # dummy message
start_time = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - start_time)
return response
@app.get("/")
async def root():
return "Wonderful!!"I made two requests with valid path ( you can see the two “200 OK” )
And, I made two request with invalid path ( two “404 Not Found” )
Based on this testing, I was able to confirm that the middleware is executed regardless the matched request path exists or not.
Experiment 3: Does the middleware work even for the non-async endpoint?
What if the API endpoint is non-async, will the middleware work?
Yes. it works!
The the middleware can be used for both async and non-async API endpoints.
import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - start_time)
return response
@app.get("/")
def root():
return "Wonderful!! - Sync"Experiment 4: What is the sequence of executions?
async def add_process_time_header(request: Request, call_next):
# 1. Do thing before the matched path operation
start_time = time.time()
# 2. find / execute the matched path operation
response = await call_next(request)
# 3. Do thing after the matched path operation
response.headers["X-Process-Time"] = str(time.time() - start_time)
return response- FastAPI ( precisely, Starlette ) will go through the middleware stacks including the user middleware. ( There are pre-defined and pre-added middleware + user middleware which is `add_process_time_header` )
- When the middleware `add_process_time_header` is executed, it will execute
start_time = time.time() - Next, in the middleware `add_process_time_header`,
response = await call_next(request)will be executed. And, the matched path,def root()route will be executed. - Next, in the middleware `add_process_time_header`,
response.headers["X-Process-Time"] = str(time.time()-start_time)is executed. - Next, return the
response
It’s pretty straightforward.
FYI: Starlette code that builds middleware stack
This is how middleware stack is built and the application is wrapped with the middlewares.
# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py
def build_middleware_stack(self) -> ASGIApp:
debug = self.debug
error_handler = None
exception_handlers: typing.Dict[
typing.Any, typing.Callable[[Request, Exception], Response]
] = {}
for key, value in self.exception_handlers.items():
if key in (500, Exception):
error_handler = value
else:
exception_handlers[key] = value
middleware = (
[Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
+ self.user_middleware
+ [
Middleware(
ExceptionMiddleware, handlers=exception_handlers, debug=debug
)
]
)
app = self.router
for cls, options in reversed(middleware):
app = cls(app=app, **options)
return appExperiment 5: ASGI Middlewares
Starlette (FastAPI is based on) is implemented based on ASGI specifications.
It means that any ASGI Middlewares ( even including third-party one ) can be used in FastAPI as well.
ASGI Middlewares are classes that expect to receive an ASGI app as the first argument.
Here is a example ASGI Middleware in Starlette. — GZipMiddleware
# ref: https://github.com/encode/starlette/blob/master/starlette/middleware/gzip.py
class GZipMiddleware:
def __init__(
self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9
) -> None:
self.app = app
self.minimum_size = minimum_size
self.compresslevel = compresslevel
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] == "http":
headers = Headers(scope=scope)
if "gzip" in headers.get("Accept-Encoding", ""):
responder = GZipResponder(
self.app, self.minimum_size, compresslevel=self.compresslevel
)
await responder(scope, receive, send)
return
await self.app(scope, receive, send)These ASGI Middlewares can be added to FastAPI application, but it should be added little different from using the @app.middleware("http")
Here is a example that shows how the ASGI Middleware — CORSMiddleware can be added into FastAPI application.
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost.tiangolo.com",
"https://localhost.tiangolo.com",
"http://localhost",
"http://localhost:8080",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)The add_middleware is Starlette’s function in application. It adds the given the middleware class into the self.user_middleware list. The added middleware class will be used when building the middleware stack and wrapping the application.
# https://github.com/encode/starlette/blob/master/starlette/applications.py
def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
if self.middleware_stack is not None: # pragma: no cover
raise RuntimeError("Cannot add middleware after an application has started")
self.user_middleware.insert(0, Middleware(middleware_class, **options))Understanding the format of add_middleware
app.add_middleware() receives a middleware class as the first argument and any additional arguments to be passed to the middleware.
app.add_middleware(
CORSMiddleware, # middleware class
# these are the keyword arguments for CORSMiddleware
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)And, here is the CORSMiddleware’s init keyword arguments.
# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py
class CORSMiddleware:
def __init__(
self,
app: ASGIApp,
allow_origins: typing.Sequence[str] = (),
allow_methods: typing.Sequence[str] = ("GET",),
allow_headers: typing.Sequence[str] = (),
allow_credentials: bool = False,
allow_origin_regex: typing.Optional[str] = None,
expose_headers: typing.Sequence[str] = (),
max_age: int = 600,
) -> None:Provided Middlewares in FastAPI and Starlette
FastAPI has a list of Middlewares. You can check from here.
Most of available Middlewares in FastAPI are originated from Starlette.
Experiment 5: How is the custom middleware added?
In the previous example, I added a custom middleware by using app.middleware("http") . This doesn’t look like the ASGI Middleware.
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
response.headers["X-Process-Time"] = str(time.time() - start_time)
return responseLet’s take a look the function, FastAPI middleware().
In FastAPI, the middleware is implemented like below. Basically, the custom middleware ( non ASGI middleware format ) is added as a part of BaseHTTPMiddleware’s dispatch function.
This is a simple way to implement and add a custom middleware without implementing the full ASGI middleware from scratch.
# ref: https://github.com/tiangolo/fastapi/blob/master/fastapi/applications.py
def middleware(
self, middleware_type: str
) -> Callable[[DecoratedCallable], DecoratedCallable]:
def decorator(func: DecoratedCallable) -> DecoratedCallable:
# this add_middleware is in Starlette
# This is the format of adding ASGI middleware as you saw before.
# BaseHTTPMiddleware is defined in Starlette
# ref: https://github.com/encode/starlette/blob/master/starlette/middleware/base.py
self.add_middleware(BaseHTTPMiddleware, dispatch=func)
return func
return decoratorThe custom middleware warpped by BaseHTTPMiddleware is added into self.user_middleware . Later the middlewares in self.user_middleware are used to build the full application. ( FYI, it is lazy-loading )
# ref: https://github.com/encode/starlette/blob/master/starlette/applications.py
def add_middleware(self, middleware_class: type, **options: typing.Any) -> None:
if self.middleware_stack is not None: # pragma: no cover
raise RuntimeError("Cannot add middleware after an application has started")
self.user_middleware.insert(0, Middleware(middleware_class, **options))Extra: Is it possible to add Middleware only to specific Router?
No, currently ( 2023–08–11), it is not supported yet. ( Precisely, there is no clean way, but there are workarounds. )
There is a open ( on-going ) PR by the FastAPI author which can support the feature adding Middleware on Router level.
Extra: What are the workarounds using Middleware only in specific Router?
There are two ways to achieve to add ( or only affect ) Middleware on Router level.
- Building a custom ASGI middleware that filter by route.
- OR creating multiple FastAPI application and mount them into one like below link
Extra: How to modify response body in Middleware?
I thought it’s super simple, but it seems not.
This linked post shows how to implement ASGI Middleware that modify response body.
