sflib.py

sflib.py

#!/usr/bin/env python3

from decimal import Decimal
import os
import unittest

import snowflake.connector


def connect(user, password, account, database):
    return snowflake.connector.connect(
        user=user,
        password=password,
        account=account,
        database=database
    )

def close(ctx):
    if ctx is not None:
        ctx.close()
        ctx = None
        return True
    return False

def bind_params(query, params):
    if params is None:
        return query

    for p in params:
        if isinstance(params[p], int) or isinstance(params[p], float):
            v = "{0}".format(params[p])
        else:
            v = "'{0}'".format(params[p].replace("'", "").replace(';', ''))
        query = query.replace(':{0}:'.format(p), v)
    return query

def query(ctx, sql):
    cs = ctx.cursor()

    rs = []
    try:
        cs.execute(sql)
        cols = []
        for c in cs.description:
            cols.append(c[0].lower())
        for r in cs.fetchall():
            d = {}
            for a in zip(cols, r):
                d[a[0]] = a[1]
            rs.append(d)
    except snowflake.connector.errors.ProgrammingError as e:
        raise Exception('{0} ({1})'.format(e, sql))
    finally:
        cs.close()

    return rs