from sqlalchemy import exc
from sqlalchemy import event
from sqlalchemy.pool import Pool
@event.listens_for(Pool, "checkout")
def ping_connection(dbapi_connection, connection_record, connection_proxy):
cursor = dbapi_connection.cursor()
try:
cursor.execute("SELECT 1")
except:
# optional - dispose the whole pool
# instead of invalidating one at a time
# connection_proxy._pool.dispose()
# raise DisconnectionError - pool will try
# connecting again up to three times before raising.
raise exc.DisconnectionError()
cursor.close()
DBアクセスの前に起こすデコレータを書く。
参考: FlaskとSQLAlchemyを使っててMySQL server has gone awayってなる - I/O Error : My Knowledge