"""DB Connector. Manages interactions with Snowflake.""" from contextlib import contextmanager from functools import wraps from oto import response import sentry_sdk from snowflake.sqlalchemy import URL 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 conflict_manager import config BaseModel = declarative_base() def _create_engine(): """Create engine based on configuration settings. Args: db_url (str): database URL Returns: obj: db_engine """ db_url = URL( account=config.SNOWFLAKE_ACCOUNT, user=config.SNOWFLAKE_USER, role=config.SNOWFLAKE_ROLE, warehouse=config.SNOWFLAKE_WAREHOUSE) if config.POOL_CLASS != pool.QueuePool: return create_engine( db_url, connect_args=config.SNOWFLAKE_CONNECT_ARGS, poolclass=config.POOL_CLASS) db_engine = create_engine( db_url, connect_args=config.SNOWFLAKE_CONNECT_ARGS, pool_size=config.POOL_SIZE, max_overflow=config.POOL_MAX_OVERFLOW, pool_recycle=config.POOL_RECYCLE) # SQLAlchemy 1.2b supports pessimistic disconnect handling out of box. # Once version 1.2 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.listen(db_engine, 'engine_connect', _ping_connection) return db_engine 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: connection.should_close_with_result = save_should_close_with_result @contextmanager def db_session(): """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) """ _db_engine = _create_engine() _db_session = sessionmaker(bind=_db_engine) session = _db_session() try: yield session session.commit() except: # noqa: E722 session.rollback() raise finally: session.close() def db_session_wrap(func): """DB Session Wrappper. Creates a new session if one isn't passed in. """ @wraps(func) def wrapper(*args, **kwargs): session = kwargs.pop('session', None) try: if session: result = func(*args, session=session, **kwargs) else: with db_session() as session: result = func(*args, session=session, **kwargs) return result except Exception as e: if config.SENTRY_DSN: # Capture exceptions using Sentry when available sentry_sdk.capture_exception(e) return response.create_fatal_response(str(e)) raise return wrapper