Skip to content

Instantly share code, notes, and snippets.

@hsuyuming
Last active June 8, 2019 09:46
Show Gist options
  • Save hsuyuming/2b191dd3951bb59422d1c6bc0b89ee68 to your computer and use it in GitHub Desktop.
Save hsuyuming/2b191dd3951bb59422d1c6bc0b89ee68 to your computer and use it in GitHub Desktop.
#import snowflake connector module
import snowflake.connector
import sys
import os
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.hazmat.primitives import serialization
# 確定輸入的參數是否給定正確
# Sample: python create_table_Insert_data.py <account> <user> <role>
if len(sys.argv) < 4 :
print("ERROR: Please pass the following command-line parameters in order:",end='\n')
print("account,user,role.")
sys.exit(-1)
else:
ACCOUNT = sys.argv[1]
USER = sys.argv[2]
ROLE = sys.argv[3]
with open("/Users/abehsu/Documents/Snowflake/Snowpipe_poc/rsa_key.p8", "rb") as key:
p_key = serialization.load_pem_private_key(
key.read(),
password = os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
backend=default_backend()
)
pkb = p_key.private_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
con = snowflake.connector.connect(
account=ACCOUNT,
user=USER,
role=ROLE,
private_key=pkb
)
results = con.cursor().execute("select current_warehouse(), current_database(), current_schema();")
for result in results:
print("current_warehouse","current_database","current_schema")
print(result[0],result[1],result[2])
con.cursor().execute("""
USE WAREHOUSE tiny_warehouse_mg;
""")
con.cursor().execute("""
USE SCHEMA testdb_mg.testschema_mg;
""")
con.cursor().execute("""
CREATE OR REPLACE TABLE test_table( col1 integer, col2 string);
""")
con.cursor().execute("""
INSERT INTO test_table(col1,col2) VALUES
(123,'test string1'),
(456,'test string2')
""")
results = con.cursor().execute("select current_warehouse(), current_database(), current_schema();")
for result in results:
print("current_warehouse","current_database","current_schema")
print(result[0],result[1],result[2])
cur = con.cursor()
try:
cur.execute("""
SELECT * FROM test_table
""")
for (col1,col2) in cur:
print("col1","col2")
print('%s,%s' %(col1,col2))
finally:
cur.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment