"""mysql connector.""" from contextlib import contextmanager from functools import wraps import time from flask import g from sqlalchemy import create_engine from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import pool from sqlalchemy import select from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from content_review import config from content_review.constants import mysql as mysql_consts def _create_engine(db_url): """Create engine based on configuration settings.""" if config.POOL_CLASS == pool.QueuePool: db_engine = create_engine( db_url, pool_size=config.POOL_SIZE, max_overflow=config.POOL_MAX_OVERFLOW, pool_recycle=config.POOL_RECYCLE_MS) # SQLAlchemy 1.12b supports pessimistic disconnect handling out of box. # Once version 1.12 is out of beta, it is recommended to use # the `pool_pre_ping` param for `create_engine` and remove any code # in this file related to `_ping_connection` # See http://docs.sqlalchemy.org/en/latest/core/pooling.html if config.POOL_PRE_PING: @event.listens_for(db_engine, 'engine_connect') def ping_connection(connection, branch): _ping_connection(connection, branch) return db_engine else: return create_engine(db_url, poolclass=config.POOL_CLASS) # Do not use these variables directly other than running unit tests db_engine = _create_engine(config.DB_URL) # please don't use sessions directly; # instead use db_session sessions = { mysql_consts.OWS_CONTENT_REVIEW: sessionmaker(bind=db_engine) } BaseModel = declarative_base() @contextmanager def db_session(db_name=mysql_consts.OWS_CONTENT_REVIEW): """Provide a transactional scope around a series of operations. Taken from http://docs.sqlalchemy.org/en/latest/orm/session_basics.html. This handles rollback and closing of session, so there is no need to do that throughout the code. Args: db_name (session): Name of database to connect to Usage: with db_session() as session: session.execute(query) """ session = sessions[db_name]() try: yield session session.commit() except: session.rollback() raise finally: session.close() def _ping_connection(connection, branch): """Ping database connection after engine_connect event. This function is copied verbatim from http://docs.sqlalchemy.org/en/latest/core/pooling.html """ if branch: # "branch" refers to a sub-connection of a connection, # we don't want to bother pinging on these. return # turn off "close with result". This flag is only used with # "connectionless" execution, otherwise will be False in any case save_should_close_with_result = connection.should_close_with_result connection.should_close_with_result = False try: # run a SELECT 1. use a core select() so that # the SELECT of a scalar value without a table is # appropriately formatted for the backend connection.scalar(select([1])) except exc.DBAPIError as err: # catch SQLAlchemy's DBAPIError, which is a wrapper # for the DBAPI's exception. It includes a .connection_invalidated # attribute which specifies if this connection is a "disconnect" # condition, which is based on inspection of the original exception # by the dialect in use. if err.connection_invalidated: # run the same SELECT again - the connection will re-validate # itself and establish a new connection. The disconnect detection # here also causes the whole connection pool to be invalidated # so that all stale connections are discarded. connection.scalar(select([1])) else: raise finally: # restore "close with result" connection.should_close_with_result = save_should_close_with_result def db_session_wrap(fn): """DB Session Wrappper. Creates a new session if one isn't passed in. """ @wraps(fn) def wrapper(*args, **kwargs): session = kwargs.pop('session', None) if session: return fn(*args, session=session, **kwargs) else: with db_session() as session: return fn(*args, session=session, **kwargs) return wrapper def db_session_retry_wrap(fn): """DB Session Wrappper with retry. Creates a new session if one isn't passed in, and does 1 retry if an internal error occurs while executing query. db_session_wrap is still preferred since this can rollback the session. """ @wraps(fn) def wrapper(*args, **kwargs): session = kwargs.pop('session', None) try: if session: return fn(*args, session=session, **kwargs) else: with db_session() as session: return fn(*args, session=session, **kwargs) except exc.InternalError as e: # Log warning and sleep for a bit g.log.warning(e) time.sleep(mysql_consts.INTERNAL_ERROR_SLEEP_TIME) if session: session.rollback() return fn(*args, session=session, **kwargs) else: with db_session() as session: return fn(*args, session=session, **kwargs) return wrapper