I have a rather inelegant solution that gets the job done for table creation and appends, all without leaving my Jupyter.
I keep this code in my sql utility file. The get_col_types function will create a dictionary of col names and dtypes needed to create the table.
def get_col_types(df):
'''
Helper function to create/modify Snowflake tables; gets the column and dtype pair for each item in the dataframe
args:
df: dataframe to evaluate
'''
import numpy as np
# get dtypes and convert to df
ct = df.dtypes.reset_index().rename(columns={0:'col'})
ct = ct.apply(lambda x: x.astype(str).str.upper()) # case matching as snowflake needs it in uppers
# only considers objects at this point
# only considers objects and ints at this point
ct['col'] = np.where(ct['col']=='OBJECT', 'VARCHAR', ct['col'])
ct['col'] = np.where(ct['col'].str.contains('DATE'), 'DATETIME', ct['col'])
ct['col'] = np.where(ct['col'].str.contains('INT'), 'NUMERIC', ct['col'])
ct['col'] = np.where(ct['col'].str.contains('FLOAT'), 'FLOAT', ct['col'])
# get the column dtype pair
l = []
for index, row in ct.iterrows():
l.append(row['index'] + ' ' + row['col'])
string = ', '.join(l) # convert from list to a string object
string = string.strip()
return string
def create_table(table, action, col_type, df):
'''
Function to create/replace and append to tables in Snowflake
args:
table: name of the table to create/modify
action: whether do the initial create/replace or appending; key to control logic
col_type: string with column name associated dtype, each pair separated by a comma; comes from get_col_types() func
df: dataframe to load
dependencies: function get_col_types(); helper function to get the col and dtypes to create a table
'''
import pandas as pd
import snowflake.connector as snow
from snowflake.connector.pandas_tools import write_pandas
from snowflake.connector.pandas_tools import pd_writer
database=database
warehouse=warehouse
schema=schema
# set up connection
conn = snow.connect(
account = ACCOUNT,
user = USER,
password = PW,
warehouse = warehouse,
database = database,
schema = schema,
role = ROLE)
# set up cursor
cur = conn.cursor()
if action=='create_replace':
# set up execute
cur.execute(
""" CREATE OR REPLACE TABLE
""" + table +"""(""" + col_type + """)""")
#prep to ensure proper case
df.columns = [col.upper() for col in df.columns]
# write df to table
write_pandas(conn, df, table.upper())
elif action=='append':
# convert to a string list of tuples
df = str(list(df.itertuples(index=False, name=None)))
# get rid of the list elements so it is a string tuple list
df = df.replace('[','').replace(']','')
# set up execute
cur.execute(
""" INSERT INTO """ + table + """
VALUES """ + df + """
""")
Working example:
# create df
l1 = ['cats','dogs','frogs']
l2 = [10, 20, 30]
df = pd.DataFrame(zip(l1,l2), columns=['type','age'])
col_type = get_col_types(df)
create_table('table_test', 'create_replace', col_type, df)
# now that the table is created, append to it
l1 = ['cow','cricket']
l2 = [45, 20]
df2 = pd.DataFrame(zip(l1,l2), columns=['type','age'])
append_table('table_test', 'append', None, df2)
#create table if not exists in schema sql = create_table_string cur.execute(sql) print('Table created successfully in Snowflake') #Truncate and write to table sql = "TRUNCATE IF EXISTS <table_name>;" cur.execute(sql) print('Truncate table created successfully in Snowflake') write_pandas(conn, df2, " <table_name>") print('Table written successfully in Snowflake') # Close your cursor and your connection. cur.close() conn.close()Everything works up until the write_pandas function, where I get Insufficient privileges to operate on schema <schema_name> – Halfway