Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_from_db_params_not_registered():
"""
Tests helpful error message on attempt to choose unregistered db_params
type.
"""
db_params = MagicMock(DbParams)
db_params.dbtype = 'Not a real type'
with pytest.raises(ETLHelperHelperError,
match=r'Unsupported DbParams.dbtype.*'):
DB_HELPER_FACTORY.from_db_params(db_params)
def test_from_dbparams(dbtype_keyword, expected_helper):
"""
Tests correct helper produced given a db params object
"""
db_params = MagicMock(DbParams)
db_params.dbtype = dbtype_keyword
helper = DB_HELPER_FACTORY.from_db_params(db_params)
assert isinstance(helper, expected_helper)
def test_from_db_params_bad_type():
with pytest.raises(ETLHelperHelperError,
match=r'Expected DbParams-like object.*'):
DB_HELPER_FACTORY.from_db_params('some string')
def connect(db_params, password_variable=None, **kwargs):
"""
Return database connection.
:param db_params: DbParams object or similar with appropriate attributes
:param password_variable: str, name of environment variable with password
:param kwargs: connection specific keyword arguments e.g. row_factory
:return: Connection object
"""
helper = DB_HELPER_FACTORY.from_db_params(db_params)
# Helpers will raise ETLHelperConnectionError if connection fails
conn = helper.connect(db_params, password_variable, **kwargs)
return conn
def get_connection_string(db_params, password_variable):
"""
Get a connection string
:param db_params: DbParams object or similar with appropriate attributes
:param password_variable: str, name of environment variable with password
:return: str, Connection string
"""
helper = DB_HELPER_FACTORY.from_db_params(db_params)
return helper.get_connection_string(db_params, password_variable)
def get_sqlalchemy_connection_string(db_params, password_variable):
"""
Get a SQLAlchemy connection string.
:param db_params: DbParams object or similar with appropriate attributes
:param password_variable: str, name of environment variable with password
:return: str, Connection string
"""
helper = DB_HELPER_FACTORY.from_db_params(db_params)
return helper.get_sqlalchemy_connection_string(db_params, password_variable)