Snowflake & CI/CD

My case for CI/CD in Snowflake is very specific: every 30 days I register for a new Snowflake 30 day trial, to run several experiments. When spinning up this new environment, it contains only the bare necessities: some roles (ACCOUNTADMIN, SYSADMIN, etc) and some databases (DEMO_DB plus the shared SNOWFLAKE databases).

So, my experiment can now start. It contains of a lot of SQL scripts that I need every time in my new Snowflake account.

For this I have chosen the most basic approach I can think of. Some people would call it an MVP :)

It consist of:

• a Python script
• a folder with SQL scripts
• a JSON file with Snowflake credentials

When you run the Python script, it will open the json, set up a connection with Snowflake:

def get_conn(): 
    with open('../config/sf-config.json','r') as file: 
        data = json.load(file) 
        user = data['user'] 
        password = data['password'] 
        account = data['account'] 
 
    conn = snowflake.connector.connect( 
        user = user, 
        password = password, 
        account = account 
        )    
    return conn

Then, the folder is browsed, sorted in alphabetical order:

def execute_sql_scripts(con): 
    cur = con.cursor() 
    for root, dirnames, filenames in os.walk("./sql"): 
             filenames = sorted(filenames) 
             for filename in fnmatch.filter(filenames, '*.sql'): 
                    sql = open(root+ "/"+ filename,'r') 
                    for cur in con.execute_stream(sql): 
                        for ret in cur: 
                            print(ret) 
    sql.close() 
    cur.close()

The rest of the script contains the flow to call the functions:

class TestSQLScriptExecution(unittest.TestCase): 
    def test_execute_sql_scripts(self): 
        con = get_conn() 
        execute_sql_scripts(con) 
        con.close() 
 
def main(): 
    con = get_conn(); 
    execute_sql_scripts(con) 
    con.close() 
 
if __name__ == "__main__": 
    print(datetime.datetime.now()) 
    unittest.main() 
    print(datetime.datetime.now())

We also need to so some imports

import snowflake.connector 
import json 
import os 
import fnmatch 
import datetime 
import unittest 
import tracemalloc 
tracemalloc.start()

The result then is that my DDL and DML is executed !!! For me, that’s enough. Don’t run this script in a production setup. When you need that, there are plenty of other options. Even Snowflake itself is entering the CI/CD arena with their Git Integration with is now in PrPr (Private Preview):

So, now about the Python script. A lot of comments can be made about it. I am not a Python programmer but use the art of “copy-paste from Stackoverflow”. Added to this script was some Bing Chat Enterprise/ ChatGpt. So, no sophisticated error handling or other Python necessities in this script.

Links:

Snowflake 30 day trial

Announcements by Snowflake