Timing Middleware
Source module: fastapi_utils.timing
¶
The fastapi_utils.timing
module provides basic profiling functionality that could be
used to find performance bottlenecks, monitor for regressions, etc.
There are currently two public functions provided by this module:
-
add_timing_middleware
, which can be used to add a middleware to aFastAPI
app that will log very basic profiling information for each request (with low overhead). -
record_timing
, which can be called on astarlette.requests.Request
instance for aFastAPI
app with the timing middleware installed (viaadd_timing_middleware
), and will emit performance information for the request at the point at which it is called.
Tip
If you are look for more fine-grained performance profiling data, consider
yappi
,
a python profiling library that was recently updated with coroutine support to enable
better coroutine-aware profiling.
Note however that yappi
adds considerable runtime overhead, and should typically be used during
development rather than production.
The middleware provided in this package is intended to be sufficiently performant for production use.
Adding timing middleware¶
The add_timing_middleware
function takes the following arguments:
app: FastAPI
: The app to which to add the timing middlewarerecord: Optional[Callable[[str], None]] = None
: The callable to call on the generated timing messages. If not provided, defaults toprint
; a good choice is theinfo
method of alogging.Logger
instanceprefix: str = ""
: A prefix to prepend to the generated route names. This can be useful for, e.g., distinguishing between mounted ASGI apps.exclude: Optional[str] = None
: If provided, any route whose generated name includes this value will not have its timing stats recorded.
Here’s an example demonstrating what the logged output looks like (note that the commented output has been split to multiple lines for ease of reading here, but each timing record is actually a single line):
import asyncio
import logging
from fastapi import FastAPI
from starlette.requests import Request
from starlette.staticfiles import StaticFiles
from starlette.testclient import TestClient
from fastapi_utils.timing import add_timing_middleware, record_timing
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
add_timing_middleware(app, record=logger.info, prefix="app", exclude="untimed")
static_files_app = StaticFiles(directory=".")
app.mount(path="/static", app=static_files_app, name="static")
@app.get("/timed")
async def get_timed() -> None:
await asyncio.sleep(0.05)
@app.get("/untimed")
async def get_untimed() -> None:
await asyncio.sleep(0.1)
@app.get("/timed-intermediate")
async def get_with_intermediate_timing(request: Request) -> None:
await asyncio.sleep(0.1)
record_timing(request, note="halfway")
await asyncio.sleep(0.1)
TestClient(app).get("/timed")
# INFO:__main__:TIMING: Wall: 53.0ms
# | CPU: 1.2ms
# | app.__main__.get_timed
TestClient(app).get("/untimed")
# <nothing logged>
TestClient(app).get("/timed-intermediate")
# INFO:__main__:TIMING: Wall: 105.3ms
# | CPU: 0.4ms
# | app.__main__.get_with_intermediate_timing (halfway)
# INFO:__main__:TIMING: Wall: 206.7ms
# | CPU: 1.1ms
# | app.__main__.get_with_intermediate_timing
TestClient(app).get("/static/test")
# INFO:__main__:TIMING: Wall: 1.6ms
# | CPU: 1.6ms
# | StaticFiles<'static'>
Recording intermediate timings¶
In the above example, you can see the get_with_intermediate_timing
function used in
the /timed-intermediate
endpoint to record an intermediate execution duration:
import asyncio
import logging
from fastapi import FastAPI
from starlette.requests import Request
from starlette.staticfiles import StaticFiles
from starlette.testclient import TestClient
from fastapi_utils.timing import add_timing_middleware, record_timing
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
add_timing_middleware(app, record=logger.info, prefix="app", exclude="untimed")
static_files_app = StaticFiles(directory=".")
app.mount(path="/static", app=static_files_app, name="static")
@app.get("/timed")
async def get_timed() -> None:
await asyncio.sleep(0.05)
@app.get("/untimed")
async def get_untimed() -> None:
await asyncio.sleep(0.1)
@app.get("/timed-intermediate")
async def get_with_intermediate_timing(request: Request) -> None:
await asyncio.sleep(0.1)
record_timing(request, note="halfway")
await asyncio.sleep(0.1)
TestClient(app).get("/timed")
# INFO:__main__:TIMING: Wall: 53.0ms
# | CPU: 1.2ms
# | app.__main__.get_timed
TestClient(app).get("/untimed")
# <nothing logged>
TestClient(app).get("/timed-intermediate")
# INFO:__main__:TIMING: Wall: 105.3ms
# | CPU: 0.4ms
# | app.__main__.get_with_intermediate_timing (halfway)
# INFO:__main__:TIMING: Wall: 206.7ms
# | CPU: 1.1ms
# | app.__main__.get_with_intermediate_timing
TestClient(app).get("/static/test")
# INFO:__main__:TIMING: Wall: 1.6ms
# | CPU: 1.6ms
# | StaticFiles<'static'>
Note that this requires the app that generated the Request
instance to have had the timing middleware
added using the add_timing_middleware
function.
This can be used to output multiple records at distinct times in order to introspect the relative contributions of different execution steps in a single endpoint.