Aiogram+SQLAlchemy+MySQL的TG机器人连接超时问题求助

问题背景 #

使用aiogram、SQLAlchemy和MySQL开发7×24运行的Telegram机器人，数据库闲置数小时后出现连接断开问题。确认是MySQL的8小时超时机制导致，添加了pool_timeout=7、pool_recycle=60、pool_pre_ping=True、isolation_level="AUTOCOMMIT"等参数后，问题仍未解决。

报错信息 #

首次触发的错误：

sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2013, 'Lost connection to MySQL server during query')

后续再次操作时触发：

exception=PendingRollbackError("Can't reconnect until invalid transaction is rolled back. Please rollback() fully before proceeding")

复现步骤 #

• 运行测试脚本
• 脚本设置了5秒超时，按回车键发送查询请求
• 等待5秒后再次按回车，出现连接丢失错误
• 再次按回车，触发PendingRollbackError错误

问题根源分析 #

1. 直接使用engine连接而非Session管理：测试代码中直接调用engine.connect()获取连接并执行操作，该连接不受SQLAlchemy连接池的pool_recycle、pool_pre_ping等配置管理，超时后无法自动回收或重连。
2. 异常捕获逻辑错误：except OperationalError or PendingRollbackError写法无效，无法同时捕获两个异常类型；且异常处理中仅回滚了Session，但实际执行操作的是独立的connection，上下文不匹配。
3. Session与connection混用：代码中同时创建了Session和独立connection，两者属于不同的连接上下文，导致连接池配置无法统一生效。

解决方案 #

1. 统一使用Session管理数据库操作：所有数据库操作都通过SQLAlchemy Session执行，让连接池配置完全生效。
2. 调整连接池参数：确保pool_recycle的值小于MySQL的wait_timeout（例如MySQL默认8小时=28800秒，可设置pool_recycle=28000），配合pool_pre_ping=True让连接池在分配连接前自动检查有效性。
3. 修正异常处理逻辑：正确捕获多个异常类型，在连接出错时回滚并关闭当前Session，重新创建新Session执行后续操作。

修改后的测试代码 #

import traceback
from sqlalchemy import create_engine, URL, text
from sqlalchemy.exc import OperationalError, PendingRollbackError
from sqlalchemy.orm import sessionmaker, declarative_base, scoped_session

from config import DB_DRIVER_NAME, DB_USERNAME, DB_HOST, DB_PASSWORD, DB_DATABASE

Base = declarative_base()

url = URL.create(
    drivername=DB_DRIVER_NAME,
    username=DB_USERNAME,
    host=DB_HOST,
    password=DB_PASSWORD,
    database=DB_DATABASE
)

# 配置连接池，pool_recycle小于MySQL的wait_timeout，这里测试用3秒
engine = create_engine(
    url,
    pool_recycle=3,
    pool_pre_ping=True,
    isolation_level="AUTOCOMMIT"
)

# 初始化Session工厂
Session = scoped_session(sessionmaker(bind=engine))

def init_test_timeout():
    # 用Session初始化测试用的短超时
    session = Session()
    try:
        session.execute(text("SET wait_timeout=5"))
        session.execute(text("SELECT 1"))
        print("测试超时设置完成")
    finally:
        session.close()

def run():
    session = Session()
    try:
        input("Press enter for sending SELECT 1 query")
        print("sql execution start")
        # 通过Session执行查询
        session.execute(text("SELECT 1"))
        print("sql execution end")
    except (OperationalError, PendingRollbackError):
        print("捕获连接错误，执行回滚并重新创建Session")
        traceback.print_exc()
        session.rollback()
        session.close()
        # 重新获取新Session继续执行
        run()
    finally:
        session.close()

if __name__ == '__main__':
    init_test_timeout()
    try:
        run()
    except Exception as e:
        print("全局异常捕获")
        traceback.print_exc()

关键优化点 #

• 所有数据库操作通过Session执行，连接池配置完全生效
• 异常处理中正确回滚并关闭出错的Session，避免无效事务残留
• pool_pre_ping=True会在每次从连接池获取连接前发送测试查询，确保连接有效
• pool_recycle定期回收超时连接，避免MySQL主动断开后连接池仍持有无效连接

内容的提问来源于stack exchange，提问作者Fəqan Çələbizadə