codekingpro/portable-devtools

1from __future__ import annotations2 3import dataclasses4import inspect5import typing as t6from time import perf_counter7 8import sqlalchemy as sa9import sqlalchemy.event as sa_event10from flask import current_app11from flask import g12from flask import has_app_context13 14 15def get_recorded_queries() -> list[_QueryInfo]:16    """Get the list of recorded query information for the current session. Queries are17    recorded if the config :data:`.SQLALCHEMY_RECORD_QUERIES` is enabled.18 19    Each query info object has the following attributes:20 21    ``statement``22        The string of SQL generated by SQLAlchemy with parameter placeholders.23    ``parameters``24        The parameters sent with the SQL statement.25    ``start_time`` / ``end_time``26        Timing info about when the query started execution and when the results where27        returned. Accuracy and value depends on the operating system.28    ``duration``29        The time the query took in seconds.30    ``location``31        A string description of where in your application code the query was executed.32        This may not be possible to calculate, and the format is not stable.33 34    .. versionchanged:: 3.035        Renamed from ``get_debug_queries``.36 37    .. versionchanged:: 3.038        The info object is a dataclass instead of a tuple.39 40    .. versionchanged:: 3.041        The info object attribute ``context`` is renamed to ``location``.42 43    .. versionchanged:: 3.044        Not enabled automatically in debug or testing mode.45    """46    return g.get("_sqlalchemy_queries", [])  # type: ignore[no-any-return]47 48 49@dataclasses.dataclass50class _QueryInfo:51    """Information about an executed query. Returned by :func:`get_recorded_queries`.52 53    .. versionchanged:: 3.054        Renamed from ``_DebugQueryTuple``.55 56    .. versionchanged:: 3.057        Changed to a dataclass instead of a tuple.58 59    .. versionchanged:: 3.060        ``context`` is renamed to ``location``.61    """62 63    statement: str | None64    parameters: t.Any65    start_time: float66    end_time: float67    location: str68 69    @property70    def duration(self) -> float:71        return self.end_time - self.start_time72 73 74def _listen(engine: sa.engine.Engine) -> None:75    sa_event.listen(engine, "before_cursor_execute", _record_start, named=True)76    sa_event.listen(engine, "after_cursor_execute", _record_end, named=True)77 78 79def _record_start(context: sa.engine.ExecutionContext, **kwargs: t.Any) -> None:80    if not has_app_context():81        return82 83    context._fsa_start_time = perf_counter()  # type: ignore[attr-defined]84 85 86def _record_end(context: sa.engine.ExecutionContext, **kwargs: t.Any) -> None:87    if not has_app_context():88        return89 90    if "_sqlalchemy_queries" not in g:91        g._sqlalchemy_queries = []92 93    import_top = current_app.import_name.partition(".")[0]94    import_dot = f"{import_top}."95    frame = inspect.currentframe()96 97    while frame:98        name = frame.f_globals.get("__name__")99 100        if name and (name == import_top or name.startswith(import_dot)):101            code = frame.f_code102            location = f"{code.co_filename}:{frame.f_lineno} ({code.co_name})"103            break104 105        frame = frame.f_back106    else:107        location = "<unknown>"108 109    g._sqlalchemy_queries.append(110        _QueryInfo(111            statement=context.statement,112            parameters=context.parameters,113            start_time=context._fsa_start_time,  # type: ignore[attr-defined]114            end_time=perf_counter(),115            location=location,116        )117    )118