Skip to content

SQL panels

SQLAlchemy

Add the SQLAlchemyPanel to your panel list:

from debug_toolbar.middleware import DebugToolbarMiddleware
from fastapi import FastAPI

app = FastAPI(debug=True)

app.add_middleware(
    DebugToolbarMiddleware,
    panels=["debug_toolbar.panels.sqlalchemy.SQLAlchemyPanel"],
)

SQLAlchemy panel

This panel records all queries using the "Dependency Injection" system as described in the FastAPI docs.

If you don't use dependencies then create a new class that inherits from SQLAlchemyPanel, override the add_engines method and add the class path to your panel list:

from debug_toolbar.panels.sqlalchemy import SQLAlchemyPanel as BasePanel
from sqlalchemy import create_engine

engine = create_engine("sqlite://", connect_args={"check_same_thread": False})


class SQLAlchemyPanel(BasePanel):
    async def add_engines(self, request: Request):
        self.engines.add(engine)

debug_toolbar.panels.sqlalchemy.SQLAlchemyPanel

Source code in debug_toolbar/panels/sqlalchemy.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class SQLAlchemyPanel(SQLPanel):
    title = "SQLAlchemy"

    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
        super().__init__(*args, **kwargs)
        self.engines: t.Set[Engine] = set()

    def register(self, engine: Engine) -> None:
        event.listen(engine, "before_cursor_execute", self.before_execute)
        event.listen(engine, "after_cursor_execute", self.after_execute)

    def unregister(self, engine: Engine) -> None:
        event.remove(engine, "before_cursor_execute", self.before_execute)
        event.remove(engine, "after_cursor_execute", self.after_execute)

    def before_execute(
        self,
        conn: Connection,
        cursor: t.Any,
        statement: str,
        parameters: t.Union[t.Sequence, t.Dict],
        context: DefaultExecutionContext,
        executemany: bool,
    ) -> None:
        conn.info.setdefault("start_time", []).append(perf_counter())

    def after_execute(
        self,
        conn: Connection,
        cursor: t.Any,
        statement: str,
        parameters: t.Union[t.Sequence, t.Dict],
        context: DefaultExecutionContext,
        executemany: bool,
    ) -> None:
        query = {
            "duration": (perf_counter() - conn.info["start_time"].pop(-1)) * 1000,
            "sql": statement,
            "params": parameters,
            "is_select": context.invoked_statement.is_select,
        }
        self.add_query(str(conn.engine.url), query)

    async def add_engines(self, request: Request):
        route = request["route"]

        if hasattr(route, "dependant"):
            if "fastapi_astack" not in request:
                async with AsyncExitStack() as stack:
                    request.scope["fastapi_astack"] = stack

            solved_result = await solve_dependencies(
                request=request,
                dependant=route.dependant,
                dependency_overrides_provider=route.dependency_overrides_provider,
            )
            for value in solved_result[0].values():
                if isinstance(value, Session):
                    self.engines.add(value.get_bind())

    async def process_request(self, request: Request) -> Response:
        await self.add_engines(request)

        for engine in self.engines:
            self.register(engine)
        try:
            response = await super().process_request(request)
        finally:
            for engine in self.engines:
                self.unregister(engine)
        return response

add_engines(request) async

Source code in debug_toolbar/panels/sqlalchemy.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def add_engines(self, request: Request):
    route = request["route"]

    if hasattr(route, "dependant"):
        if "fastapi_astack" not in request:
            async with AsyncExitStack() as stack:
                request.scope["fastapi_astack"] = stack

        solved_result = await solve_dependencies(
            request=request,
            dependant=route.dependant,
            dependency_overrides_provider=route.dependency_overrides_provider,
        )
        for value in solved_result[0].values():
            if isinstance(value, Session):
                self.engines.add(value.get_bind())

Tortoise ORM

Add the TortoisePanel to your panel list:

from debug_toolbar.middleware import DebugToolbarMiddleware
from fastapi import FastAPI

app = FastAPI(debug=True)

app.add_middleware(
    DebugToolbarMiddleware,
    panels=["debug_toolbar.panels.tortoise.TortoisePanel"],
)