Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ##### Initialize elixir/SQLAlchemy
- # Disconnect handling
- 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):
- logging.debug("***********ping_connection**************")
- cursor = dbapi_connection.cursor()
- try:
- cursor.execute("SELECT 1")
- except:
- logging.debug("######## DISCONNECTION ERROR #########")
- # 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()
- metadata.bind= create_engine("mysql://foo:bar@localhost/db_name", pool_size=100, pool_recycle=3600)
- setup_all()
Add Comment
Please, Sign In to add comment