Instrumenting and Observing Async Queries

Attach before_cursor_execute and after_cursor_execute to the sync engine behind your AsyncEngine, record the elapsed time and the statement, and you have per-query timing that costs microseconds and works everywhere — which is the foundation every other form of database observability is built on. This page sits under async engines, dialects and connection pooling; start there for the engine and pool model these hooks attach to.

Concept & Execution Model

An AsyncEngine is a thin wrapper around a synchronous engine driven through the greenlet bridge. Event listeners attach to that inner engine, reached with engine.sync_engine, and they run inside the greenlet — synchronously, on the event loop thread, once per statement.

That has two consequences worth internalising. Listeners must not await, because they are synchronous callbacks; anything asynchronous has to be handed to a queue or a background task. And whatever they do costs wall-clock time on every query, so a listener that takes a millisecond has just added a millisecond to every statement the application issues.

# instrumentation.py — per-query timing, the foundation everything else builds on
from __future__ import annotations

import logging
import time

from sqlalchemy import event
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.ext.asyncio import AsyncEngine

logger = logging.getLogger("db.query")
SLOW_QUERY_SECONDS = 0.25


def install_query_timing(async_engine: AsyncEngine) -> None:
    """Attach timing hooks to the synchronous engine behind an AsyncEngine."""
    engine: Engine = async_engine.sync_engine

    @event.listens_for(engine, "before_cursor_execute")
    def _before(conn: Connection, cursor, statement, parameters, context, executemany):
        # Per-connection, so concurrent coroutines cannot overwrite each other:
        # each has its own connection for the duration of its statement.
        conn.info["query_started"] = time.perf_counter()

    @event.listens_for(engine, "after_cursor_execute")
    def _after(conn: Connection, cursor, statement, parameters, context, executemany):
        started = conn.info.pop("query_started", None)
        if started is None:
            return
        elapsed = time.perf_counter() - started
        if elapsed >= SLOW_QUERY_SECONDS:
            logger.warning(
                "slow query",
                extra={
                    "elapsed_ms": round(elapsed * 1000, 1),
                    # The statement, never the parameters: those are user data.
                    "statement": " ".join(statement.split())[:500],
                    "param_count": len(parameters) if parameters else 0,
                    "executemany": executemany,
                },
            )

Storing the start time in conn.info rather than in a module-level variable is the detail that makes this correct under concurrency. Each statement executes on its own connection, and conn.info is per-connection, so two coroutines running simultaneously cannot overwrite each other's timestamps.

Query Construction & Async Execution Patterns

Counting queries per unit of work is the cheapest high-value instrument, because it turns an N+1 from a production mystery into a number you can assert on.

# Async — a per-request query counter built on a ContextVar
from __future__ import annotations

import contextvars
from dataclasses import dataclass, field

query_stats: contextvars.ContextVar["QueryStats | None"] = contextvars.ContextVar(
    "query_stats", default=None
)


@dataclass
class QueryStats:
    count: int = 0
    total_seconds: float = 0.0
    statements: list[str] = field(default_factory=list)


def install_query_counting(async_engine: AsyncEngine) -> None:
    engine = async_engine.sync_engine

    @event.listens_for(engine, "after_cursor_execute")
    def _count(conn, cursor, statement, parameters, context, executemany):
        stats = query_stats.get()
        if stats is None:
            return
        stats.count += 1
        started = conn.info.get("query_started")
        if started is not None:
            stats.total_seconds += time.perf_counter() - started
        if len(stats.statements) < 50:            # bounded: never grow without limit
            stats.statements.append(" ".join(statement.split())[:200])

A ContextVar is per-task under asyncio, so concurrent requests each get their own counter with no locking and no interference. Wiring it into a framework is a middleware that sets it, resets it, and reports:

# Starlette / FastAPI middleware
from starlette.middleware.base import BaseHTTPMiddleware


class QueryStatsMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        token = query_stats.set(QueryStats())
        try:
            response = await call_next(request)
        finally:
            stats = query_stats.get()
            query_stats.reset(token)          # never leak into the next request
        response.headers["X-Query-Count"] = str(stats.count)
        response.headers["X-Query-Ms"] = str(round(stats.total_seconds * 1000, 1))
        if stats.count > 25:
            logger.warning("query fan-out", extra={"count": stats.count,
                                                   "path": request.url.path})
        return response

# Sync — the same counter, for a WSGI application
def wsgi_query_stats(app):
    def middleware(environ, start_response):
        token = query_stats.set(QueryStats())
        try:
            return app(environ, start_response)
        finally:
            query_stats.reset(token)
    return middleware

The reset(token) in the finally block is the part most often omitted, and its absence is the same bug described for tenant context in dynamic schema and multi-tenant routing: state set for one request bleeding into the next one handled by the same task.

State Management & Session Boundaries

Instrumentation reveals session boundaries that were previously invisible, and the numbers it produces are usually more surprising than the slow-query log.

The first surprise is normally the query count for a handler nobody suspected. A page that renders twenty orders and issues forty-one queries has an N+1, and the counter above finds it in the first request rather than in a profiler session weeks later. The remedy is a loader option, chosen by cardinality as set out in using selectinload vs joinedload for N+1 prevention.

The second is when the connection is held. Query timing measures the statements; it does not show the gap between them, which is where an external HTTP call, a template render or a slow serialiser sits while a pooled connection stays checked out. Measuring that needs a hook at the checkout level rather than the statement level:

def install_checkout_timing(async_engine: AsyncEngine) -> None:
    """How long each connection is held, which is what actually sizes the pool."""
    engine = async_engine.sync_engine

    @event.listens_for(engine, "checkout")
    def _checkout(dbapi_conn, connection_record, connection_proxy):
        connection_record.info["checked_out_at"] = time.perf_counter()

    @event.listens_for(engine, "checkin")
    def _checkin(dbapi_conn, connection_record):
        started = connection_record.info.pop("checked_out_at", None)
        if started is None:
            return
        held = time.perf_counter() - started
        if held > 1.0:
            logger.warning("connection held", extra={"held_ms": round(held * 1000)})

A connection held for a second while the queries inside it total four milliseconds is the signature of work that does not belong inside the session boundary — and it is the number that decides how large the pool has to be, as the arithmetic in tuning connection pools for cloud databases shows.

Advanced Tracing Across the Greenlet Bridge

Distributed tracing is where async SQLAlchemy needs care that a synchronous application does not. Spans propagate through a ContextVar, and the greenlet bridge runs the ORM internals in a different greenlet — which shares the context, but only because greenlet_spawn copies it. Starting a span inside after_cursor_execute and ending it elsewhere does not work; both ends must be in the same hook pair.

# OpenTelemetry spans, one per statement, correctly parented
from __future__ import annotations

from opentelemetry import trace
from opentelemetry.trace import SpanKind

tracer = trace.get_tracer("sqlalchemy")


def install_tracing(async_engine: AsyncEngine, db_name: str) -> None:
    engine = async_engine.sync_engine

    @event.listens_for(engine, "before_cursor_execute")
    def _start(conn, cursor, statement, parameters, context, executemany):
        span = tracer.start_span(
            _operation_name(statement),
            kind=SpanKind.CLIENT,
            attributes={
                "db.system": "postgresql",
                "db.name": db_name,
                # The statement text, with values already replaced by placeholders.
                "db.statement": " ".join(statement.split())[:1000],
                "db.sqlalchemy.executemany": executemany,
            },
        )
        conn.info["otel_span"] = span

    @event.listens_for(engine, "after_cursor_execute")
    def _end(conn, cursor, statement, parameters, context, executemany):
        span = conn.info.pop("otel_span", None)
        if span is not None:
            span.end()

    @event.listens_for(engine, "handle_error")
    def _error(context):
        span = context.connection.info.pop("otel_span", None) if context.connection else None
        if span is not None:
            span.record_exception(context.original_exception)
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            span.end()


def _operation_name(statement: str) -> str:
    """A low-cardinality span name: the verb and the table, never the full SQL."""
    head = statement.lstrip().split(None, 4)
    verb = head[0].upper() if head else "SQL"
    if verb in {"SELECT", "DELETE"} and "FROM" in statement.upper():
        table = statement.upper().split("FROM", 1)[1].split()[0]
    elif verb in {"INSERT", "UPDATE"} and len(head) > 2:
        table = head[2]
    else:
        table = ""
    return f"{verb} {table.strip('\"')}".strip()

The handle_error listener matters more than it looks: without it, a failing statement leaves its span open forever, and the trace shows a query that never completed rather than one that failed. It is also the hook that lets a driver error be attached to the span that caused it, which is what makes a trace useful during an incident rather than after one.

Span names must be low-cardinality — SELECT orders, not the full statement — or the tracing backend groups nothing and every query becomes its own operation. That is why _operation_name extracts a verb and a table rather than using the statement text.

Hybrid Architectures & Migration Strategies

Most applications adopt instrumentation incrementally, and the order that pays off fastest is not the order teams usually pick.

Start with the query counter, not the slow-query log. Counting is cheaper to add, cheaper to run, and catches the failure mode that actually degrades applications: not one slow query, but two hundred fast ones. It also converts directly into a test assertion, which nothing else on this list does.

# The counter earns its keep in tests, where an N+1 becomes a failure
@pytest.mark.asyncio
async def test_order_list_does_not_fan_out(session, client):
    token = query_stats.set(QueryStats())
    try:
        await client.get("/orders?limit=20")
        assert query_stats.get().count <= 3, query_stats.get().statements
    finally:
        query_stats.reset(token)

That assertion fails the moment someone removes a loader option, which is the only reliable way to keep an N+1 fixed. The rollback-isolated fixtures it depends on are covered in testing async SQLAlchemy applications.

Add timing second, with a threshold rather than a log line per query. Logging every statement is how a DEBUG log becomes the largest cost centre in the system, and echo=True — which does exactly that, parameters included — belongs in development only.

Add pool metrics third, scraped on a timer rather than per request, for the reasons set out in handling connection leaks and pool exhaustion: sampling inside a request biases every reading upward.

Add tracing last, and only when there is more than one service. In a single service, timing plus counting answers nearly every question tracing would, at a fraction of the operational cost.

Production Pitfalls & Anti-Patterns

• echo=True in production. Logs every statement and its bound parameters, which means every value the application has handled ends up in the log. Treat enabling it in a deployed environment as an incident.
• Module-level state in a listener. Two concurrent coroutines overwrite each other's start times, producing timings that are wrong in an uninteresting way — often negative. Use conn.info for per-statement state and a ContextVar for per-request state.
• RuntimeError: no running event loop inside a listener. Something tried to await or schedule a coroutine from a synchronous hook. Push the work onto a queue the application drains elsewhere.
• Unbounded statement lists. A per-request stats object that appends every statement will happily hold ten thousand strings for a batch job. Cap it, as the example above does.
• High-cardinality span or metric names. Using the full SQL as a span name or a metric label makes the backend group nothing and, for metrics, produces an unbounded cardinality explosion that is expensive in a way that is hard to reverse.
• Instrumenting the AsyncEngine instead of sync_engine. Listeners attached to the async wrapper never fire for cursor-level events. The symptom is silence, which is easy to mistake for "no queries are slow".
• Leaving query_stats set after a request. Without reset(token) in a finally, the next request handled by the same task inherits the previous one's counter, and the numbers become nonsense under load only.

Reading a Query Plan From the ORM

Timing tells you a statement is slow; it does not tell you why. For that you need the plan, and getting one from an ORM query means compiling the statement with its parameters inlined and asking the database to explain it.

from __future__ import annotations

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import Select


async def explain(session: AsyncSession, stmt: Select, analyze: bool = False) -> str:
    """Return the plan for an ORM statement, as PostgreSQL would print it.

    literal_binds inlines the parameters, which is what makes the plan reflect the
    ACTUAL values — a plan for a generic parameter can differ enormously from the
    plan for the value that is actually slow.
    """
    compiled = stmt.compile(
        dialect=session.bind.dialect,
        compile_kwargs={"literal_binds": True},
    )
    prefix = "EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)" if analyze else "EXPLAIN"
    result = await session.execute(text(f"{prefix} {compiled}"))
    return "\n".join(row[0] for row in result)

Two cautions come with it. literal_binds inlines values directly into the SQL, so this helper must never be given user input in a deployed environment — it is a development and diagnosis tool, not a request path. And ANALYZE actually runs the statement, so explaining a DELETE with analyze=True deletes the rows; wrap it in a transaction you roll back, or restrict the helper to SELECT.

What to look for in the output is usually one of three things. A sequential scan on a large table where you expected an index seek means the predicate is not index-friendly — often a function applied to the column, or a type mismatch that forces a cast. A row-count estimate wildly different from the actual count means the statistics are stale, which ANALYZE on the table fixes and which is common immediately after a bulk load. And a nested loop over many rows where a hash join was expected usually means the planner underestimated the row count, which points back at the same statistics.

Capturing plans automatically for the slowest queries is tempting and mostly a trap: the plan for a query that was slow once is rarely reproducible afterwards, because the buffers are now warm. The pragmatic arrangement is a slow-query log that records the statement, and a development helper like the one above for reproducing it deliberately.

Frequently Asked Questions

Related

• Async Engines, Dialects, and Connection Pooling — The parent guide: the engine and pool model these listeners attach to.
• Logging slow async queries with SQLAlchemy events — The timing listener in full, with thresholds, sampling and safe statement formatting.
• Counting queries per request to catch N+1 regressions — The ContextVar counter, the middleware, and turning it into a test assertion.
• Adding OpenTelemetry tracing to async SQLAlchemy — Spans that survive the greenlet bridge, and low-cardinality naming.
• Reading EXPLAIN output for a SQLAlchemy query — Turning "this statement is slow" into a plan, and the three readings that explain most of them.
• Handling connection leaks and pool exhaustion — The pool counters to scrape, and why sampling inside a request biases them.