Database utilities, including connection strings to pyodbc or turbodbc etc..

dbutils.py

import sqlalchemy
import os


def get_mssql_turbodbc_engine(server, database):
    conn = "mssql+turbodbc://{0}/{1}?trusted_connection=yes;driver=ODBC+Driver+13+for+SQL+Server".format(server, database)
    return sqlalchemy.create_engine(conn)
    
    
def get_mssql_pyodbc_engine(server, database):
    conn = "mssql+pyodbc://{0}/{1}?trusted_connection=yes;driver=SQL+Server".format(server, database)
    return sqlalchemy.create_engine(conn)
    
    
def get_drill_conn(dsn="MapR Drill", host="s-cp-rmp01.LD.corp.local", port=31010) -> pyodbc.Connection:
    dsn = os.getenv("DRILL_DSN", dsn)
    host = os.getenv("DRILL_HOST", host)
    port = os.getenv("DRILL_PORT", port)
    conn_str = "DSN={};CONNECTIONTYPE=DIRECT;HOST={};PORT={}".format(dsn, host, port)
    return pyodbc.connect(conn_str, autocommit=True)
    
    

def check_table_exists(tablename, schema='dbo', dbcon=None, engine=None):
    """
    One of dbcon or (sqlalchemy) engine should be passed
    :param tablename:
    :param schema:
    :param dbcon:
    :param engine:
    :return:
    """
    dbcon = dbcon or engine.raw_connection()
    dbcur = dbcon.cursor()
    dbcur.execute("""
        SELECT COUNT(*)
        FROM information_schema.tables
        WHERE table_name = '{0}' AND table_schema = '{1}'
        """.format(tablename.replace('\'', '\'\''), schema))
    if dbcur.fetchone()[0] == 1:
        dbcur.close()
        return True

    dbcur.close()
    return False