Middlewares¶

FastAPI (and starlette) already provides some useful middlewares however along with that I prefer some custom solution that can be used with my tech stack.

SQLAlchemy Query Count¶

This middleware returns total number of queries executed for each request via response header X-DB-Query-Count.

ExamplePreview

from fastapi import FastAPI
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from fastapi_batteries.fastapi.middlewares import QueryCountMiddleware

app = FastAPI()


engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)


app.add_middleware(QueryCountMiddleware, engine=engine)


@app.get("/")
async def get_index():
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))
        await conn.commit()

        return {"message": "Hello World"}

Request Process Time¶

This middleware returns execution time of each request via response header X-Process-Time.

ExamplePreview

from fastapi import FastAPI

from fastapi_batteries.fastapi.middlewares import RequestProcessTimeMiddleware

app = FastAPI()

app.add_middleware(RequestProcessTimeMiddleware)


@app.get("/")
async def get_index():
    return {"message": "Hello World"}