How to use the etlhelper.db_helpers.PostgresDbHelper function in etlhelper

To help you get started, we’ve selected a few etlhelper examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_postgres_connect(monkeypatch):
    db_params = DbParams(dbtype='PG',
                         host='server', port='1521', dbname='testdb',
                         user='testuser', odbc_driver='test driver')
    monkeypatch.setenv('DB_PASSWORD', 'mypassword')
    expected_conn_str = 'host=server port=1521 dbname=testdb user=testuser password=mypassword'
    mock_connect = Mock()
    monkeypatch.setattr(psycopg2, 'connect', mock_connect)

    # Act
    helper = PostgresDbHelper()
    helper.connect(db_params, 'DB_PASSWORD')

    # Assert
    mock_connect.assert_called_with(expected_conn_str)
github BritishGeologicalSurvey / etlhelper / test / unit / test_helper_factory.py View on Github external
                          (PostgresDbHelper, psycopg2.extensions.connection),
                          (MSSQLDbHelper, pyodbc.Connection)])
def test_from_conn(expected_helper, db_class):
    """
    Tests correct helper produced given a conn object
    """
    conn = Mock()
    # conn.__class__ = cx_Oracle.Connection
    conn.__class__ = db_class
    helper = DB_HELPER_FACTORY.from_conn(conn)
    assert isinstance(helper, expected_helper)
github BritishGeologicalSurvey / etlhelper / test / unit / test_db_helpers.py View on Github external
def test_postgres_sqlalchemy_connect(monkeypatch):
    db_params = DbParams(dbtype='PG',
                         host='server', port='1521', dbname='testdb',
                         user='testuser')
    monkeypatch.setenv('DB_PASSWORD', 'mypassword')
    helper = PostgresDbHelper()
    conn_str = helper.get_sqlalchemy_connection_string(db_params, 'DB_PASSWORD')
    expected_conn_str = 'postgresql://testuser:mypassword@server:1521/testdb'

    assert conn_str == expected_conn_str