python: sqlalchemy - 如何使用新的事件系统确保连接不会过时

我正在 python 中使用 sqlalchemy 包。在现有表上执行自动加载后，我有一个操作需要一些时间才能执行。当我尝试使用连接时，这会导致以下错误：

sqlalchemy.exc.OperationalError: (OperationalError) (2006, 'MySQL server has gone away')

我有一个简单的实用函数，可以执行多次插入：

def insert_data(data_2_insert, table_name):
    engine = create_engine('mysql://blah:blah123@localhost/dbname')
    # Metadata is a Table catalog. 
    metadata = MetaData()
    table = Table(table_name, metadata, autoload=True, autoload_with=engine)
    for c in mytable.c:
        print c
    column_names = tuple(c.name for c in mytable.c)
    final_data = [dict(zip(column_names, x)) for x in data_2_insert]
    ins = mytable.insert()
    conn = engine.connect()
    conn.execute(ins, final_data)
    conn.close()

由于“data_2_insert”有 677,161 行，因此以下行的执行时间很长。

final_data = [dict(zip(column_names, x)) for x in data_2_insert]

我碰到这个问题 https://stackoverflow.com/questions/3033234/handle-mysql-restart-in-sqlalchemy这指的是类似的问题。但是我不确定如何实现建议的连接管理接受的答案 https://stackoverflow.com/a/3034003/445073因为 robots.jpg 在评论中指出了这一点：

SQLAlchemy 0.7 的注意事项 - PoolListener 已弃用，但可以使用新的实现相同的解决方案事件系统 http://docs.sqlalchemy.org/en/latest/core/pooling.html#disconnect-handling-pessimistic.

如果有人可以向我展示一些关于如何将这些建议集成到我使用 sqlalchemy 的方式中的指示，我将非常感激。谢谢。

我想你正在寻找这样的东西：

from sqlalchemy import exc, event
from sqlalchemy.pool import Pool

@event.listens_for(Pool, "checkout")
def check_connection(dbapi_con, con_record, con_proxy):
    '''Listener for Pool checkout events that pings every connection before using.
    Implements pessimistic disconnect handling strategy. See also:
    http://docs.sqlalchemy.org/en/rel_0_8/core/pooling.html#disconnect-handling-pessimistic'''

    cursor = dbapi_con.cursor()
    try:
        cursor.execute("SELECT 1")  # could also be dbapi_con.ping(),
                                    # not sure what is better
    except exc.OperationalError, ex:
        if ex.args[0] in (2006,   # MySQL server has gone away
                          2013,   # Lost connection to MySQL server during query
                          2055):  # Lost connection to MySQL server at '%s', system error: %d
            # caught by pool, which will retry with a new connection
            raise exc.DisconnectionError()
        else:
            raise

如果您希望有条件地触发此策略，您应该避免在此处使用装饰器，而是使用注册侦听器listen()功能：

# somewhere during app initialization
if config.check_connection_on_checkout:
    event.listen(Pool, "checkout", check_connection)

更多信息：

• 连接池事件 http://docs.sqlalchemy.org/en/rel_0_8/core/event.html
• 事件API http://docs.sqlalchemy.org/en/rel_0_8/core/events.html