Fixing "garbage collector is trying to clean up non-checked-in connection" warnings

The warning means a connection was checked out and never returned — a session, AsyncConnection or streamed result that was not closed — so wrap every one in async with, and add checkout tracking to find the call site that is not. This guide belongs to handling connection leaks and pool exhaustion.

Quick Answer

The full message under asyncpg reads: The garbage collector is trying to clean up non-checked-in connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7f…>>, which will be dropped, as it cannot be safely terminated. Please ensure that SQLAlchemy pooled connections are returned to the pool explicitly, either by calling close() or by using appropriate context managers to manage their lifecycle.

It is not a driver problem. Something checked a connection out and nothing gave it back.

Before — a helper that opens a session and never closes it:

from sqlalchemy import select

from shop.db import Session
from shop.models import Customer


async def find_customer(email: str) -> Customer | None:
    session = Session()                     # checked out on first execute
    customer = await session.scalar(select(Customer).where(Customer.email == email))
    if customer is None:
        return None                         # early return: session never closed
    await session.close()
    return customer

After — the session's lifetime is a block:

from sqlalchemy import select

from shop.db import Session
from shop.models import Customer


async def find_customer(email: str) -> Customer | None:
    async with Session() as session:
        return await session.scalar(select(Customer).where(Customer.email == email))

The async with form closes the session on every path — normal return, early return, exception and task cancellation — which is why it is the only form worth using. The same applies to engine.connect(), engine.begin() and session.stream().

Execution Context & Async Workflow Integration

SQLAlchemy's pool hands out connections wrapped in a small proxy object and counts them as checked out until the proxy is returned. Returning is explicit: closing a session releases its connection, closing an AsyncConnection releases it, and exiting their context managers closes them. If none of that happens, the proxy stays alive as long as something references the session — and when the last reference goes, Python's garbage collector eventually finalises it.

At that point the pool notices a connection that was never checked in. With a synchronous driver it can close the connection right there. With asyncpg it cannot: closing an asyncpg connection is a coroutine, and a finaliser running inside the garbage collector cannot await anything or be sure an event loop is even running. So SQLAlchemy drops the connection object, logs the warning, and adjusts the pool's count. PostgreSQL sees the socket disappear eventually and ends the backend.

Three consequences are worth understanding. First, the warning arrives late and at an arbitrary place — during garbage collection, possibly in a different request, possibly minutes after the leak — so its stack trace says nothing about the cause. Second, until the collector runs, the leaked connection counts against pool_size + max_overflow, so leaks show up first as QueuePool limit reached timeouts and only later, if at all, as this warning. Third, any transaction the connection had open is neither committed nor cleanly rolled back from the application's side; the server rolls it back when the connection finally dies, holding its locks until then.

Two situations produce the warning without a real leak in application code. At interpreter shutdown, objects are collected in an arbitrary order after the event loop has closed, so a service that exits without await engine.dispose() can log it for connections that were merely idle in sessions still referenced from module globals. And a test suite that creates engines per test without disposing them produces a stream of these warnings as fixtures go out of scope. Both are fixed by disposing engines deliberately, shown below.

Resolving Warnings, Errors & Common Mistakes

Streams deserve a specific note because the leak is not obvious. session.stream() returns an AsyncResult backed by a server-side cursor on a checked-out connection. Iterating it to the end releases the cursor, but breaking out of the loop early leaves it open until the result is closed:

from sqlalchemy import select

from shop.db import Session
from shop.models import Order


async def first_large_order(threshold_cents: int) -> Order | None:
    async with Session() as session:
        async with (await session.stream_scalars(
            select(Order).order_by(Order.id).execution_options(yield_per=500)
        )) as orders:
            async for order in orders:
                if order.total_cents >= threshold_cents:
                    return order      # the inner async with closes the cursor
    return None

Cancellation is the last common source. A task cancelled between checkout and close only cleans up if the close is in a context manager or a finally block. Code that calls await session.close() as its last line leaks on every timeout. Streaming patterns in general are covered in using yield_per to stream millions of rows.

Advanced: Finding the Leak With Checkout Tracking

When a codebase has hundreds of places that open sessions, reading them all is slower than asking the pool which one leaked. Pool events make that straightforward: record a short stack trace when a connection is checked out, discard it when it is checked in, and periodically report anything that has been out for too long.

from __future__ import annotations

import asyncio
import logging
import time
import traceback

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

log = logging.getLogger("db.leaks")


def install_checkout_tracking(engine: AsyncEngine, *, max_age_s: float = 60.0,
                              every_s: float = 30.0) -> asyncio.Task:
    pool = engine.sync_engine.pool
    outstanding: dict[int, tuple[float, str]] = {}

    @event.listens_for(pool, "checkout")
    def _checkout(dbapi_conn, record, proxy):
        stack = "".join(traceback.format_stack(limit=12)[:-2])
        outstanding[id(record)] = (time.monotonic(), stack)

    @event.listens_for(pool, "checkin")
    def _checkin(dbapi_conn, record):
        outstanding.pop(id(record), None)

    @event.listens_for(pool, "invalidate")
    def _invalidate(dbapi_conn, record, exception):
        outstanding.pop(id(record), None)

    async def report() -> None:
        while True:
            await asyncio.sleep(every_s)
            now = time.monotonic()
            for started, stack in list(outstanding.values()):
                if now - started > max_age_s:
                    log.warning(
                        "connection checked out for %.0fs", now - started,
                        extra={"checkout_stack": stack},
                    )

    return asyncio.create_task(report(), name="db-checkout-report")

Run it in staging, or in production behind a flag for a short window: traceback.format_stack() on every checkout costs tens of microseconds, which is noticeable on a hot path. The report names the function that checked out the connection, and almost always that function, or its caller, is missing an async with.

The age threshold should sit above your longest legitimate checkout. A report streaming for five minutes is not a leak; a request handler holding a connection for sixty seconds almost certainly is. Pairing this with the metrics from instrumenting and observing async queries — checked-out connections over time — shows whether a fix actually stopped the climb.

Finally, shut down cleanly so the warning stops being noise at exit. In FastAPI, dispose the engine in the lifespan; in scripts, dispose inside the coroutine passed to asyncio.run():

import asyncio

from shop.db import engine
from shop.jobs import rebuild_search_index


async def main() -> None:
    try:
        await rebuild_search_index()
    finally:
        await engine.dispose()   # closes pooled connections while the loop is alive


asyncio.run(main())

Keeping Leaks From Coming Back

A leak fixed once tends to return with the next helper someone writes in a hurry. Two cheap guards catch the pattern before it ships: a test-suite assertion that nothing is left checked out, and a review rule for how sessions may be created.

The assertion belongs in an autouse fixture. After each test, every session the test opened should be closed, so the pool's checked-out count should be back where it started:

import gc

import pytest
import pytest_asyncio

from shop.db import engine


@pytest_asyncio.fixture(autouse=True)
async def no_leaked_connections():
    pool = engine.sync_engine.pool
    before = pool.checkedout()
    yield
    gc.collect()   # make leaked proxies visible now rather than in a later test
    leaked = pool.checkedout() - before
    assert leaked <= 0, f"{leaked} connection(s) still checked out after the test"

Calling gc.collect() before the check is deliberate. Without it, a leaked session that happens to survive until a later test is blamed on that test, which sends people looking in the wrong place. With it, the failing test is the one that leaked.

The review rule is simpler to state than to enforce: a session factory call is always the expression of an async with. Session() on its own line, assigned to a variable, is a leak waiting for an early return. A few lines of an AST check in CI turn the rule into a failure:

import ast
import pathlib
import sys

FACTORIES = {"Session", "AsyncSession", "BackgroundSession"}


def bare_session_calls(path: pathlib.Path) -> list[int]:
    tree = ast.parse(path.read_text(), filename=str(path))
    managed = {
        id(item.context_expr)
        for node in ast.walk(tree) if isinstance(node, ast.AsyncWith)
        for item in node.items
    }
    return [
        node.lineno
        for node in ast.walk(tree)
        if isinstance(node, ast.Call)
        and isinstance(node.func, ast.Name)
        and node.func.id in FACTORIES
        and id(node) not in managed
    ]


problems = [
    f"{path}:{line}" for path in pathlib.Path("shop").rglob("*.py")
    for line in bare_session_calls(path)
]
if problems:
    print("session created outside async with:
  " + "
  ".join(problems))
    sys.exit(1)

The check has false positives — a dependency that yields from async with Session() is fine and passes, but a factory passed as a callback may not — so give it an allow-list rather than weakening the rule.

Frequently Asked Questions

Related

• Handling Connection Leaks and Pool Exhaustion — The parent guide: leaks, timeouts and pool sizing under load.
• Debugging QueuePool limit reached timeouts — What a slow leak looks like before the warning appears.
• Configuring pool_pre_ping to handle stale connections — The different problem of connections the server closed.
• Running background tasks with a fresh AsyncSession in FastAPI — A common source of sessions nobody closes.