ASGI Server-Timing middleware

An ASGI middleware that wraps the excellent yappi profiler to let you measure the execution time of any function or coroutine in the context of an HTTP request, and return it as a standard Server-Timing HTTP header.

Sample configurations

Here are some example configurations for various frameworks and libraries. Feel free to combine them as needed.


fastapi_app.add_middleware(ServerTimingMiddleware, calls_to_track={
	"1deps": (fastapi.routing.solve_dependencies,),
	"2main": (fastapi.routing.run_endpoint_function,),
	"3valid": (pydantic.fields.ModelField.validate,),
	"4encode": (fastapi.encoders.jsonable_encoder,),
	"5render": (


from starlette.middleware import Middleware

middleware = [
  Middleware(ServerTimingMiddleware, calls_to_track={
	  # TODO: ...

starlette_app = Starlette(routes=routes, middleware=middleware)


fastapi_app.add_middleware(ServerTimingMiddleware, calls_to_track={
	"db_exec": (sqlalchemy.engine.base.Engine.execute,),
	"db_fetch": (

More Frameworks

Feel free to submit PRs containing examples for more libraries and ASGI frameworks.


  • Only the end-to-end time is reported for both functions and coroutines, so it's not possible to tell from the metrics when a coroutine took a long time because the event loop thread got stalled (though asyncio's debug mode can help).
  • The profiler's memory is not freed over time, and only gets cleared when it exceeds a given threshold (50MB by default). When memory gets cleared, data collected for all ongoing requests is lost, so the timing for these will be incorrect.
  • Executing the same task multiple times in parallel (such as with asyncio.gather()) will report the duration as if they had been executed sequentially.
  • The minimum version of Python supported is 3.7, since this middleware makes use of PEP 567 context variables to track which function call belongs to which request, and the Python 3.6 backport doesn't have asyncio support.

Special Thanks

  • Sümer Cip (@sumerc), for creating and maintaininng yappi, as well as being very responsive and open to adding all the new features needed to make this work.
  • David Montague (@dmontagu) for his involvement in shaping this middleware at every step of the way.
  • `pydantic.fields.ModelField.validate` doesn't work anymore

    `pydantic.fields.ModelField.validate` doesn't work anymore

    In the README, it says to use fastapi.routing.ModelField.validate in order to profile how long it takes to validate the response.

    No matter which function calls I ask it to profile I'm able to get yappi to return a profile for the 3valid step.

        "3valid": (

    I can confirm by entering a debugger that these functions are being called, it's just that yappi isn't profiling them. I'm not entirely sure why.

    I know you haven't looked in this in a while and you did so much research. to get this working in the first place. It's greatly appreciated!

    Here's a repro:

    import fastapi
    import pydantic
    from contextvars import ContextVar
    from typing import Dict, List, Tuple, Callable
    import re
    import yappi
    from fastapi import FastAPI
    from yappi import YFuncStats
    _yappi_ctx_tag: ContextVar[int] = ContextVar("_yappi_ctx_tag", default=-1)
    def _get_context_tag() -> int:
        return _yappi_ctx_tag.get()
    class ServerTimingMiddleware:
        """Timing middleware for ASGI HTTP applications
        The resulting profiler data will be returned through the standard
        `Server-Timing` header for all requests.
            app (ASGI v3 callable): An ASGI application
            calls_to_track (Dict[str,Tuple[Callable]]): A dict of functions
                keyed by desired output metric name.
                Metric names must consist of a single rfc7230 token
            max_profiler_mem (int): Memory threshold (in bytes) at which yappi's
                profiler memory gets cleared.
        .. _Server-Timing sepcification:
        def __init__(
            self, app, calls_to_track: Dict[str, Tuple[Callable]], max_profiler_mem: int = 50_000_000
            for metric_name, profiled_functions in calls_to_track.items():
                if len(metric_name) == 0:
                    raise ValueError("A Server-Timing metric name cannot be empty")
                # USASCII (7 bits), only visible characters (no non-printables or space), no double-quote or delimiter
                if (
                    not metric_name.isascii()
                    or not metric_name.isprintable()
                    or'[ "(),/:;<=>[email protected]\[\\\]{}]', metric_name) is not None
                    raise ValueError(
                        '"{}" contains an invalid character for a Server-Timing metric name'.format(
                if not all(callable(profiled) for profiled in profiled_functions):
                    raise TypeError(
                        'One of the targeted functions for key "{}" is not a function'.format(
   = app
            self.calls_to_track = {
                name: list(tracked_funcs) for name, tracked_funcs in calls_to_track.items()
            self.max_profiler_mem = max_profiler_mem
        async def __call__(self, scope, receive, send):
            ctx_tag = id(scope)
            def wrapped_send(response):
                if response["type"] == "http.response.start":
                    tracked_stats: Dict[str, YFuncStats] = {
                        name: yappi.get_func_stats(
                            filter_callback=lambda x: yappi.func_matches(x, tracked_funcs),
                        for name, tracked_funcs in self.calls_to_track.items()
                    # NOTE (sm15): Might need to be altered to account for various edge-cases
                    timing_ms = {
                        name: sum(x.ttot for x in stats) * 1000
                        for name, stats in tracked_stats.items()
                        if not stats.empty()
                    server_timing = ",".join(
                        [f"{name};dur={duration_ms:.3f}" for name, duration_ms in timing_ms.items()]
                    if server_timing:
                        # FIXME: Doesn't check if a server-timing header is already set
                        response["headers"].append([b"server-timing", server_timing])
                    if yappi.get_mem_usage() >= self.max_profiler_mem:
                return send(response)
            await, receive, wrapped_send)
    track: Dict[str, Tuple[Callable, ...]] = {
        "1deps": (fastapi.routing.solve_dependencies,),
        "2main": (fastapi.routing.run_endpoint_function,),
        "3valid": (
        "4encode": (fastapi.encoders.jsonable_encoder,),
        "5render": (
    app = FastAPI()
    app.add_middleware(ServerTimingMiddleware, calls_to_track=track)
    class Item(pydantic.main.BaseModel):
        name: int = pydantic.Field(description="name")
        val: int = pydantic.Field(description="val")
    @app.get("/", response_model=List[Item])
    def test():
        resp = [{"name": x, "val": x} for x in range(10000)]
        return resp

    And here's the result from hitting localhost:8000/ image

    opened by traviscook21 1
  • Cython functions aren't supported because of the way `inspect.isfunction` works

    Cython functions aren't supported because of the way `inspect.isfunction` works

    inspect.isfunction returns False for Cython functions (see this discussion).

    As a result, tracking pydantic.fields.ModelField.validate for pydantic==1.7.3 is impossible.

    opened by wxd 4
  • Is it published to pypi?

    Is it published to pypi?

    I don't see any instructions on how to install it from pypi, only from GitHub. Is there any plans to publish it there?

    opened by devova 0
  • Create a script

    Create a script

    I don't really have much experience in the way of packaging and publishing python modules, but that's probably something I should get around doing if I want people to be able to use this.

    opened by sm-Fifteen 3
a lightweight web framework based on fastapi

start-fastapi Version 2021, based on FastAPI, an easy-to-use web app developed upon Starlette Framework Version 2020 中文文档 Requirements python 3.6+ (fo

HiKari 61 Sep 29, 2021
ASGI middleware for authentication, rate limiting, and building CRUD endpoints.

Piccolo API Utilities for easily exposing Piccolo models as REST endpoints in ASGI apps, such as Starlette and FastAPI. Includes a bunch of useful ASG

null 37 Oct 13, 2021
Reusable utilities for FastAPI

Reusable utilities for FastAPI Documentation: Source Code: FastAPI i

David Montague 868 Oct 22, 2021
A Jupyter server based on FastAPI (Experimental)

jupyverse is experimental and should not be used in place of jupyter-server, which is the official Jupyter server.

Jupyter Server 78 Oct 18, 2021
FastAPI + Django experiment

django-fastapi-example This is an experiment to demonstrate one potential way of running FastAPI with Django. It won't be actively maintained. If you'

Jordan Eremieff 54 Sep 16, 2021
python template private service

Template for private python service This is a cookiecutter template for an internal REST API service, written in Python, inspired by layout-golang. Th

UrvanovCompany 12 Nov 24, 2020
An extension for GINO to support Starlette server.

gino-starlette Introduction An extension for GINO to support starlette server. Usage The common usage looks like this: from starlette.applications imp

GINO Community 64 Oct 18, 2021
JSON-RPC server based on fastapi

Description JSON-RPC server based on fastapi: Motivation Autogenerated OpenAPI and Swagger (thanks to fastapi) for JSON-R

null 125 Oct 14, 2021
A Prometheus Python client library for asyncio-based applications

aioprometheus aioprometheus is a Prometheus Python client library for asyncio-based applications. It provides metrics collection and serving capabilit

null 90 Oct 20, 2021
Auth for use with FastAPI

FastAPI Auth Pluggable auth for use with FastAPI Supports OAuth2 Password Flow Uses JWT access and refresh tokens 100% mypy and test coverage Supports

David Montague 59 Oct 6, 2021
A rate limiter for Starlette and FastAPI

SlowApi A rate limiting library for Starlette and FastAPI adapted from flask-limiter. Note: this is alpha quality code still, the API may change, and

Laurent Savaete 308 Oct 21, 2021
FastAPI framework plugins

Plugins for FastAPI framework, high performance, easy to learn, fast to code, ready for production fastapi-plugins FastAPI framework plugins Cache Mem

RES 163 Oct 14, 2021
Web Inventory tool, takes screenshots of webpages using Pyppeteer (headless Chrome/Chromium) and provides some extra bells & whistles to make life easier.

WitnessMe WitnessMe is primarily a Web Inventory tool inspired by Eyewitness, its also written to be extensible allowing you to create custom function

byt3bl33d3r 525 Oct 22, 2021
sample web application built with FastAPI + uvicorn

SPARKY Sample web application built with FastAPI & Python 3.8 shows simple Flask-like structure with a Bootstrap template index.html also has a backgr

mrx 19 Jun 16, 2021
Deploy an inference API on AWS (EC2) using FastAPI Docker and Github Actions

Deploy an inference API on AWS (EC2) using FastAPI Docker and Github Actions To learn more about this project: medium blog post The goal of this proje

Ahmed BESBES 26 Sep 13, 2021
FastAPI Skeleton App to serve machine learning models production-ready.

FastAPI Model Server Skeleton Serving machine learning models production-ready, fast, easy and secure powered by the great FastAPI by Sebastián Ramíre

null 208 Oct 20, 2021
Monitor Python applications using Spring Boot Admin

Pyctuator Monitor Python web apps using Spring Boot Admin. Pyctuator supports Flask, FastAPI, aiohttp and Tornado. Django support is planned as well.

SolarEdge Technologies 97 Oct 22, 2021

cookiecutter-pytorch-fastapi 基于Pytorch的 脚手架项目 按规范添加推理函数即可实现Celery+FastAPI+Gunicorn+Nginx+Supervisor+Docker的快速部署 Requirements Python >= 3.6 with pip in

null 12 Oct 15, 2021

FastAPI 框架,高性能,易学,快速编码,随时可供生产 文档: 源码: FastAPI 是一个现代、快速(高性能)的 Web 框架,基于标准 Python 类型提示,使用

ApacheCN 27 Jun 28, 2021