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
14
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
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, named=True)
        event.listen(engine, "after_cursor_execute", self.after_execute, named=True)

    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, context: ExecutionContext, **kwargs: t.Any) -> None:
        context._start_time = perf_counter()  # type: ignore[attr-defined]

    def after_execute(self, context: ExecutionContext, **kwargs: t.Any) -> None:
        query = {
            "duration": (
                perf_counter() - context._start_time  # type: ignore[attr-defined]
            )
            * 1000,
            "sql": context.statement,
            "params": context.parameters,
        }
        self.add_query(str(context.engine.url), query)

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

        if hasattr(route, "dependant"):
            try:
                solved_result = await solve_dependencies(
                    request=request,
                    dependant=route.dependant,
                    dependency_overrides_provider=route.dependency_overrides_provider,
                    async_exit_stack=AsyncExitStack(),
                )
            except HTTPException:
                pass
            else:
                for value in solved_result[0].values():
                    if isinstance(value, Session):
                        bind = value.get_bind()

                        if isinstance(bind, Connection):
                            self.engines.add(bind.engine)
                        else:
                            self.engines.add(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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async def add_engines(self, request: Request):
    route = request["route"]

    if hasattr(route, "dependant"):
        try:
            solved_result = await solve_dependencies(
                request=request,
                dependant=route.dependant,
                dependency_overrides_provider=route.dependency_overrides_provider,
                async_exit_stack=AsyncExitStack(),
            )
        except HTTPException:
            pass
        else:
            for value in solved_result[0].values():
                if isinstance(value, Session):
                    bind = value.get_bind()

                    if isinstance(bind, Connection):
                        self.engines.add(bind.engine)
                    else:
                        self.engines.add(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"],
)