Skip to content

Instantly share code, notes, and snippets.

@snaga
Created March 26, 2022 05:59
Show Gist options
  • Save snaga/ec64f921727cca6d277fb0cfe172676a to your computer and use it in GitHub Desktop.
Save snaga/ec64f921727cca6d277fb0cfe172676a to your computer and use it in GitHub Desktop.
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment