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 
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
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)

    def add_bind(self, bind: Connection | Engine):
        if isinstance(bind, Connection):
            self.engines.add(bind.engine)
        else:
            self.engines.add(bind)

    async def add_engines(self, request: Request):
        dependencies = await get_dependencies(request)

        if dependencies is not None:
            for value in dependencies.values():
                if isinstance(value, AsyncSession):
                    value = value.sync_session

                if isinstance(value, Session):
                    try:
                        bind = value.get_bind()
                    except UnboundExecutionError:
                        for bind in value._Session__binds.values():  # type: ignore[attr-defined]
                            self.add_bind(bind)
                    else:
                        self.add_bind(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 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def add_engines(self, request: Request):
    dependencies = await get_dependencies(request)

    if dependencies is not None:
        for value in dependencies.values():
            if isinstance(value, AsyncSession):
                value = value.sync_session

            if isinstance(value, Session):
                try:
                    bind = value.get_bind()
                except UnboundExecutionError:
                    for bind in value._Session__binds.values():  # type: ignore[attr-defined]
                        self.add_bind(bind)
                else:
                    self.add_bind(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"],
)