Logging slow async queries with SQLAlchemy events

Attach before_cursor_execute and after_cursor_execute to engine.sync_engine, stash a perf_counter() in conn.info, and log only the statements that exceed a threshold — never the bound parameters, and never every query. This is the timing instrument from instrumenting and observing async queries, in full.

Quick Answer

Before — echo=True, which is not a slow-query log:

engine = create_async_engine(DATABASE_URL, echo=True)
# INFO sqlalchemy.engine.Engine SELECT customers.email ... WHERE customers.id = %(id)s
# INFO sqlalchemy.engine.Engine [generated in 0.00012s] {'id': 4821}
#                                                       ^^^^^^^^^^^ user data, in the log

After — a threshold listener on the synchronous engine:

from __future__ import annotations

import logging
import time

from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

logger = logging.getLogger("db.slow")


def install_slow_query_log(async_engine: AsyncEngine, threshold_s: float = 0.25) -> None:
    engine = async_engine.sync_engine

    @event.listens_for(engine, "before_cursor_execute")
    def _before(conn, cursor, statement, parameters, context, executemany):
        conn.info["query_started"] = time.perf_counter()

    @event.listens_for(engine, "after_cursor_execute")
    def _after(conn, cursor, statement, parameters, context, executemany):
        started = conn.info.pop("query_started", None)
        if started is None:
            return
        elapsed = time.perf_counter() - started
        if elapsed < threshold_s:
            return
        logger.warning(
            "slow query",
            extra={
                "elapsed_ms": round(elapsed * 1000, 1),
                "statement": " ".join(statement.split())[:500],
                "param_count": _count_params(parameters),
                "executemany": bool(executemany),
            },
        )


def _count_params(parameters) -> int:
    if parameters is None:
        return 0
    if isinstance(parameters, (list, tuple)):
        return len(parameters)
    return len(parameters) if hasattr(parameters, "__len__") else 1


engine = create_async_engine(DATABASE_URL)
install_slow_query_log(engine, threshold_s=0.25)

On a healthy request this produces no output at all. During an incident it produces exactly the statements that were slow, without a single user value.

Execution Context & Async Workflow Integration

These listeners run synchronously, inside the greenlet that AsyncSession established, on the event loop thread. That places two hard constraints on what they may do.

They cannot await. A listener that needs to reach a network service — pushing a metric, calling an API — has to hand the work to something else, and the only safe handoff from synchronous code on the loop is a thread-safe queue drained elsewhere.

# Async — handing work out of a synchronous listener
import queue
import threading

_slow_queries: queue.SimpleQueue = queue.SimpleQueue()


def _after(conn, cursor, statement, parameters, context, executemany):
    started = conn.info.pop("query_started", None)
    if started is None:
        return
    elapsed = time.perf_counter() - started
    if elapsed >= 0.25:
        # Non-blocking: the listener returns immediately, and a worker does the work.
        _slow_queries.put_nowait((elapsed, " ".join(statement.split())[:500]))


def _drain_forever() -> None:
    while True:
        elapsed, statement = _slow_queries.get()
        metrics.observe("db_slow_query_seconds", elapsed, statement=statement)


threading.Thread(target=_drain_forever, daemon=True).start()

And whatever they do costs wall-clock time on every statement. A listener that formats a statement, walks a parameter structure, or serialises anything is adding that cost to the fast queries too — which is why the example above does its formatting only after the threshold check has passed.

The measurement itself covers driver execution: network out, server work, network back. It excludes statement compilation, which happens before before_cursor_execute, and result hydration, which happens after after_cursor_execute returns. A handler that is slow while every statement is fast is spending its time in one of those two places, or between statements — and finding that needs the checkout-level timing described in the parent guide, not this listener.

Resolving Warnings, Errors & Common Mistakes

Advanced: Sampling, Rate Limiting and Statement Fingerprints

A threshold is enough until the database itself degrades — at which point every statement crosses it and the log becomes the second problem. Two additions keep it useful under exactly the conditions you need it.

from __future__ import annotations

import hashlib
import random
import re
import time
from collections import defaultdict

# Collapse literals so `WHERE id = 4821` and `WHERE id = 9130` share a fingerprint.
_NUMBER = re.compile(r"\b\d+\b")
_QUOTED = re.compile(r"'[^']*'")
_IN_LIST = re.compile(r"IN\s*\([^)]*\)", re.IGNORECASE)


def fingerprint(statement: str) -> str:
    normalised = " ".join(statement.split())
    normalised = _IN_LIST.sub("IN (?)", normalised)
    normalised = _QUOTED.sub("?", normalised)
    normalised = _NUMBER.sub("?", normalised)
    return hashlib.sha1(normalised.encode()).hexdigest()[:12]


class SlowQueryReporter:
    """Threshold, then sample, then rate-limit per fingerprint."""

    def __init__(self, threshold_s: float = 0.25, sample: float = 1.0,
                 per_fingerprint_per_minute: int = 6) -> None:
        self.threshold_s = threshold_s
        self.sample = sample
        self.limit = per_fingerprint_per_minute
        self._seen: dict[str, list[float]] = defaultdict(list)

    def should_report(self, elapsed: float, statement: str) -> str | None:
        if elapsed < self.threshold_s:
            return None
        if self.sample < 1.0 and random.random() > self.sample:
            return None
        fp = fingerprint(statement)
        now = time.monotonic()
        recent = [t for t in self._seen[fp] if now - t < 60.0]
        if len(recent) >= self.limit:
            self._seen[fp] = recent
            return None
        recent.append(now)
        self._seen[fp] = recent
        return fp

The fingerprint is what makes the rate limit meaningful: without it, a thousand distinct-but-identical statements each get their own budget and the flood returns. With it, a single badly-planned query reports six times a minute and the rest of the log stays readable.

Sampling is worth setting below one only on very high-throughput services, and it changes what the log means — an absent statement no longer implies an absent problem. A useful compromise is to sample the reporting but count every occurrence in a metric, so the counter stays exact while the log stays bounded.

The other addition worth making is a second, much higher threshold that always logs and never samples. A query taking ten seconds is qualitatively different from one taking three hundred milliseconds, and it should never be dropped by a sampling decision.

Choosing a Threshold You Will Actually Act On

A threshold set by intuition is usually either so low that the log is noise or so high that nothing ever appears. Setting it from the data takes one afternoon and does not need to be repeated often.

Start by recording elapsed time for every query as a histogram metric — which is cheap, because a histogram observation is arithmetic — and leave the log threshold high enough to be silent. After a day of real traffic, look at the distribution: the p99 is a reasonable first threshold, because by construction it fires on one query in a hundred.

Then adjust for what you can act on. If the p99 is four milliseconds because the workload is all primary-key lookups, a threshold there will fire constantly on queries nobody can improve. If it is two seconds because a nightly report dominates, a threshold there will miss everything the API does. In both cases the answer is a per-context threshold rather than a single global one:

# A lower threshold for request-path queries than for batch work
from myapp.context import in_request_path

def _threshold() -> float:
    return 0.15 if in_request_path() else 2.0

Finally, decide in advance what a log line means operationally. If a slow query does not lead to anyone looking at it, the threshold is set for a purpose nobody has, and the honest move is either to alert on the metric instead or to raise the threshold until the log is short enough that someone reads it. A slow-query log that is never read is indistinguishable from one that is never written, except in cost.

Frequently Asked Questions

Related

• Instrumenting and Observing Async Queries — The parent guide: counting, pool metrics, tracing and reading a query plan.
• Counting queries per request to catch N+1 regressions — The instrument that catches two hundred fast queries, which a slow-query log never will.
• Adding OpenTelemetry tracing to async SQLAlchemy — Connecting a slow statement to the request that issued it, across services.
• Handling connection leaks and pool exhaustion — When the slow part is waiting for a connection rather than running a query.