如何通过SQLAlchemy减少PostgreSQL连接的空闲时间

我有一款基于FastAPI与SQLAlchemy构建的应用，所有数据库相关操作均通过该技术栈处理。

当前端触发API调用时，我观察到数据库连接建立后会长时间保持IDLE状态，是否可以通过SQLAlchemy配置缩短连接的空闲时长？

我当前使用如下代码连接PostgreSQL数据库：

import sqlalchemy as db

eng = db.create_engine(<SQLALCHEMY_DATABASE_URI>)
conn = eng.connect()
metadata = db.MetaData()
table = db.Table(
            <table_name>,
            metadata,
            autoload=True,
            autoload_with=eng)
user_id = 1
try:
    if ids_by_user is None:
        query = db.select([table.columns.created_at]).where(
                table.columns.user_id == user_id,
            ).order_by(
                table.columns.created_at.desc()
            )
        result = conn.execute(query).fetchmany(1)
        
        time = result[0][0]
        time_filtering_query = db.select([table]).where(
            table.columns.created_at == time
        )
        time_result = conn.execute(time_filtering_query).fetchall()
        conn.close()
        return time_result
    else:
        output_by_id = []
        for i in ids_by_user:
            query = db.select([table]).where(
                db.and_(
                    table.columns.id == i,
                    table.columns.user_id == user_id
                )
            )
            result = conn.execute(query).fetchall()
            output_by_id.append(result)

        output_by_id = [output_by_id[j][0]
                        for j in range(len(output_by_id))
                        if output_by_id[j]]
        conn.close()
        return output_by_id
finally:
    eng.dispose()

即使用户退出登录应用，相关连接仍会在一段时间内保持活跃空闲状态，无法立即关闭。

补充说明 #

我曾尝试使用NullPool配置，但连接依旧处于空闲、ROLLBACK状态，效果与未配置NullPool时完全一致。

问题核心原因 #

连接长时间处于IDLE/ROLLBACK状态本质是连接生命周期管理逻辑错误，和连接池本身配置关系不大，之前配置NullPool无效也是同一原因：

• 当前代码每次请求都新建engine实例、手动获取连接，完全违背SQLAlchemy设计逻辑：engine是应用全局单例，应在启动时初始化一次，本身负责管理整个连接池，反复创建销毁反而会导致连接清理逻辑异常
• 执行查询后从未显式结束事务：SQLAlchemy 1.x版本默认开启隐式事务，哪怕是只读查询，只要没有显式调用commit()/rollback()，事务就会一直挂起，就算调用conn.close()，只要连接没有被真正断开，PostgreSQL端就会一直显示连接处于idle in transaction (ROLLBACK)状态
• 表结构反射（autoload）放在请求逻辑里，每次请求都重复加载表元数据，还会额外占用连接资源

具体修复方案 #

1. 全局初始化Engine和连接池配置

engine只在应用启动时初始化一次，通过连接池参数控制空闲连接回收时长：

import sqlalchemy as db
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends

app = FastAPI()
# 替换为实际数据库地址
SQLALCHEMY_DATABASE_URI = "postgresql://用户名:密码@数据库地址:端口/库名"

# 全局单例engine，只初始化一次
eng = db.create_engine(
    SQLALCHEMY_DATABASE_URI,
    pool_pre_ping=True, # 取连接时先检测可用性，避免死连接
    pool_size=5, # 连接池常驻连接数，不需要高并发可设为0
    max_overflow=10, # 最大额外临时连接数
    pool_recycle=30, # 连接空闲超过30秒自动回收断开，可按需调整时长
    pool_timeout=5, # 获取连接超时时间
    # 完全不需要连接复用、每次请求都新建断开连接，就取消下一行注释，同时注释掉所有pool_*开头的参数
    # poolclass=db.NullPool,
)

# 应用启动时一次性加载表元数据，不要每次请求重复加载
metadata = db.MetaData()
table = db.Table(
    "你的表名",
    metadata,
    autoload=True,
    autoload_with=eng
)

# 初始化Session工厂，用Session代替直接操作裸连接，自动处理事务逻辑
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=eng)

2. 用FastAPI依赖注入管理连接生命周期

不要在接口逻辑里手动开关连接，通过依赖自动处理连接的获取、关闭、事务收尾：

# 数据库连接依赖
def get_db():
    session = SessionLocal()
    try:
        yield session
    finally:
        # 请求结束时自动关闭session，未提交的事务会自动回滚，不会残留ROLLBACK状态
        session.close()

3. 重构接口查询逻辑

去掉手动创建engine、手动开关连接、手动dispose的逻辑，通过注入的session执行查询：

@app.get("/你的接口路径")
def query_data(user_id: int = 1, ids_by_user: list | None = None, session = Depends(get_db)):
    if ids_by_user is None:
        query = db.select([table.columns.created_at]).where(
            table.columns.user_id == user_id,
        ).order_by(
            table.columns.created_at.desc()
        )
        result = session.execute(query).fetchmany(1)
        if not result:
            return []
        time = result[0][0]
        time_filtering_query = db.select([table]).where(
            table.columns.created_at == time
        )
        time_result = session.execute(time_filtering_query).fetchall()
        return [dict(row._mapping) for row in time_result]
    else:
        output_by_id = []
        for i in ids_by_user:
            query = db.select([table]).where(
                db.and_(
                    table.columns.id == i,
                    table.columns.user_id == user_id
                )
            )
            result = session.execute(query).fetchall()
            if result:
                output_by_id.append(dict(result[0]._mapping))
        return output_by_id

补充说明 #

数据库端短时间显示IDLE状态是正常现象：如果使用连接池，连接回收会按配置的pool_recycle周期执行，到时间会自动断开；如果使用NullPool，请求结束时连接会直接断开，不会残留空闲连接。之前配置NullPool无效，就是因为隐式事务没有被正确收尾，连接根本没有被正常关闭。

内容的提问来源于stack exchange，提问作者some_programmer