连接池泄漏首先是所有权问题：业务工作已经结束，但它借走的连接仍然处于借出状态。此时数据库可能仍然正常响应其他客户端，只有这个应用拿不到连接。应把归还资源视为操作契约的一部分，覆盖异常和提前返回，而不是只在成功路径末尾顺手清理一下。

一份借用应当有明确的主人和终点

SQLAlchemy Connection 表示从 Engine 所属池中取得的一次借用。关闭这个 Connection，通常是把底层 DBAPI 连接还回池里，而不一定真正断开数据库连接。归还借用与提交事务是两个独立决定：池可以运转正常，但业务修改已经回滚；池也可能耗尽，而底层数据库连接全部仍然存活。只检查连接是否活着，无法回答谁还占着它。

资源作用域应尽量短，但不能破坏事务要求的原子性。with engine.begin() 在代码块成功时提交，失败时回滚，并释放连接；with engine.connect() 管理的是连接作用域，不承诺自动提交写入，关闭时尚未结束的事务会被回滚。两者看起来都用了 with，但持久化语义不同，不能为了“统一写法”随意互换，否则资源不再泄漏了，数据却可能没有保存。

与当前事务无关的远程等待、用户交互，不应长时间占着事务和连接。这并不意味着可以把必须一起提交的变更拆开。若连接由调用方拥有，辅助函数不应私自关闭；若辅助函数自己借用，就应负责归还，或者在接口中明确把所有权交给谁。只有打开次数的指标，却说不清每份借用由谁结束，仍然没有定位真正的生命周期边界。

坏掉的样子

漏掉一次归还，就减少一份可用容量，后来者开始排队或超时。对于配置了有限容量和获取超时的 QueuePool，新调用并非一定永远挂住，而是可能在执行 SQL 之前就收到池超时。把上限调大可以暂时推迟复发，却也增加数据库压力；缺失的归还路径依然存在，不能算修复。

不过，“池满了”不是泄漏的充分证据。真实并发、长事务、数据库锁等待或应用自身死锁，都可能占住全部借用。数据库 CPU 很低也不决定原因，因为应用可能握着连接等待别的事情。需要把借出数量、获取等待、持有年龄、事务状态和仍在进行的业务工作关联起来，判断这些连接是否还处于合理的使用范围。

一个有说服力的泄漏复现，应证明已经结束的工作仍然保留借用，并且归还这些具体借用后容量恢复。不要依赖垃圾回收及时释放；对象何时被回收不是可靠的资源归还契约。Engine.dispose() 也不能强制收回仍然借出的连接，它处理的是池内已经空闲的连接。重启进程可能暂时清空症状，但不会改变下一次请求重新走进同一遗漏分支的事实。

快速验证

在 macOS 或 Linux 的空目录运行 python3 -m venv .venv，然后执行 .venv/bin/python -m pip install SQLAlchemy==2.0.52。把下面保存为 pool_probe.py，执行 .venv/bin/python pool_probe.py。本次使用 Python 3.14.4 和 SQLite 3.53.0，数据库文件只存在于脚本自己的临时目录，结束后自动清理，不会访问生产数据库。

from pathlib import Path
from tempfile import TemporaryDirectory

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.exc import TimeoutError as PoolTimeout
from sqlalchemy.pool import QueuePool


class WorkFailed(Exception):
    pass


assert sqlalchemy.__version__ == "2.0.52"
with TemporaryDirectory(prefix="pool-leak-") as directory:
    engine = create_engine(
        "sqlite:///" + str(Path(directory) / "probe.db"),
        poolclass=QueuePool, pool_size=2, max_overflow=0, pool_timeout=0.05,
    )
    borrowed = []
    try:
        with engine.begin() as connection:
            connection.exec_driver_sql("CREATE TABLE ledger(id INTEGER PRIMARY KEY)")
        for _ in range(2):
            borrowed.append(engine.connect())  # Deliberately retain both leases.
        assert engine.pool.checkedout() == 2
        try:
            unexpected = engine.connect()
        except PoolTimeout:
            pass
        else:
            unexpected.close()
            raise AssertionError("retained leases must exhaust the two-slot pool")
        assert engine.pool.checkedout() == 2
        print("two retained leases exhaust the pool; the next checkout times out")

        borrowed[0].close()
        with engine.begin() as connection:
            connection.exec_driver_sql("INSERT INTO ledger VALUES (1)")
        assert engine.pool.checkedout() == 1
        print("returning one lease restores capacity without enlarging the pool")

        try:
            with engine.begin() as connection:
                connection.exec_driver_sql("INSERT INTO ledger VALUES (2)")
                raise WorkFailed("injected after the insert")
        except WorkFailed:
            pass
        assert engine.pool.checkedout() == 1
        with engine.connect() as connection:
            assert connection.exec_driver_sql("SELECT id FROM ledger").scalars().all() == [1]
            connection.exec_driver_sql("INSERT INTO ledger VALUES (3)")
            # No commit: this connection context rolls back when it closes.
        with engine.connect() as connection:
            assert connection.exec_driver_sql("SELECT id FROM ledger").scalars().all() == [1]
        print("an exception rolls back row 2 and returns its lease; closing without commit also removes row 3")

        def early_return():
            with engine.connect() as connection:
                return connection.exec_driver_sql("SELECT 1").scalar_one()

        assert early_return() == 1 and engine.pool.checkedout() == 1
        with engine.begin() as connection:
            connection.exec_driver_sql("INSERT INTO ledger VALUES (4)")
        with engine.connect() as connection:
            rows = connection.exec_driver_sql("SELECT id FROM ledger ORDER BY id").scalars().all()
        assert rows == [1, 4] and engine.pool.checkedout() == 1
        borrowed[1].close()
        assert engine.pool.checkedout() == 0 and engine.pool.checkedin() == 2
        print("early return releases its lease; successful transaction commits row 4; final rows", rows)
    finally:
        for connection in borrowed:
            connection.close()
        engine.dispose()

列表刻意保留两条借用的强引用，使模拟泄漏不依赖垃圾回收是否碰巧及时运行。第三次获取必须抛出 PoolTimeout。归还第一条后，不扩大池也能插入并提交编号一的记录，证明恢复来自资源归还，而不是新增连接预算。此时另一条故意保留的借用仍然存在，所以每次正常作用域结束后，预期借出计数是一，而不是零。

接着在插入编号二之后注入 WorkFailed。异常离开事务上下文，插入被回滚，借用也归还。另一段连接上下文插入编号三，却没有 commit；关闭连接同样回滚这条记录。提前 return 的函数也仍然执行上下文退出，因此不会绕过归还。最后用成功事务提交编号四，排序后的查询结果必须严格等于 [1, 4]，同时证明哪些路径提交、哪些路径回滚，而不只是打印“执行成功”。

关闭最后一条故意保留的借用后，借出数量降为零，池内仍有两条可复用的空闲物理连接。最后的清理先归还所有保留句柄，再 dispose Engine。这些断言验证的是本地池计数和事务结果，不是生产锁竞争、取消传播或分布式副作用；归还连接不会撤销已经发送到其他系统的 HTTP 调用。真正修改容量前，应把同样的所有权检查放到业务实际失败路径上，再决定问题是泄漏、持有过久还是容量确实不足。

来源