Snowflake DDL Comparator(Different Accounts) using Streamlit

“Managing object definition between different environments is a common requirement in software development when dealing with databases. Using Streamlit to visualize and manage this process can streamline the workflow and make it more user-friendly.”

Below, we’ll explore how to achieve the same functionality using Streamlit.

Part1: Set Up Connection & Cursor

import snowflake.connector
import streamlit as st
import pandas as pd

# Connect to Snowflake accounts


conn_account1 = snowflake.connector.connect(
    user='account1username',
    password='account1pwd',
    account='account1name',
    warehouse='activeWHnameaccount1'
)


def get_snowflake_connection(sfid):
    return snowflake.connector.connect(
    user=account2username,
    password='account2pwd',
    authenticator='externalbrowser',(only when SSO auth is required)
    account='account2name',
    warehouse='activeWHnameaccount2'

    )

# Connect to Snowflake
sfid = 'account2username'
conn_account2 = get_snowflake_connection(sfid)

#Define Cursors:

sfcur_account1=conn_account1.cursor()
sfcur_account2=conn_account2.cursor()

Part2: Set up Function for Object DDL

# Function to retrieve DDL statement for a table or view
def get_object_ddl(cursor, database_name, schema_name, object_name, object_type):
    cursor = cursor
    try:
        cursor.execute(f'SELECT GET_DDL(\'{object_type.upper()}\', \'{database_name.upper()}.{schema_name.upper()}.{object_name.upper()}\')')
        row = cursor.fetchone()
        if row:
            return row[0]
        else:
            return f'{object_type.capitalize()} not found'
    except Exception as e:
        return f'Error: {str(e)}'

Part3: Streamlit UI related parameters and Object DDL related params:

# Streamlit app
st.title('Snowflake DDL Comparator')


database_name=st.text_input('Enter Database Name')
schema_name = st.text_input('Enter Schema Name')

object_type = st.selectbox('Select Object Type', ['Table', 'View'])

num_objects = st.text_input(f'Enter Number of {object_type}s')

if num_objects.isdigit() and int(num_objects) > 0:
    num_objects = int(num_objects)
    object_names = []
    for i in range(num_objects):
        object_names.append(st.text_input(f'Enter {object_type} Name {i+1}'))

    if st.button('Compare DDLs'):
        data = []
        for object_name in object_names:
            ddl_account_account1 = get_object_ddl(sfcur_account1, database_name, schema_name, object_name,object_type)
            ddl_account_account2 = get_object_ddl(sfcur_account2, database_name, schema_name, object_name,object_type)

            match_status = 'Match' if ddl_account_account1 == ddl_account_account2 else 'Not Matched'
            data.append([object_name, ddl_account_account1, ddl_account_account2, match_status])

        df = pd.DataFrame(data, columns=['Object Name', 'Account1 DDL', 'Account2 DDL', 'Match Status'])
        st.write(df)

Part4: Close the cursors :

sfcur_account1.close()
sfcur_account2.close()

Once Completed : Save the File with .py extension and run the file using streamlit in conda env.

More Details Here Related to Execution: https://medium.com/@peggie7191/streamlit-with-snowflake-27485af200d4

Execution StepWise:

1: Input Database Name, Schema Name and Number of Same object type to be compared.

2: Select Object Type (Table/View) from dropdown menu.

3. Mention Object Names One by One.

4: Click on "Compare DDL" button to initiate comparison step.

5: Match or Unmatch Status as per check in the accounts is displayed across each object name.

UI Display :

After providing all the inputs and clicking on ‘Compare DDLs’, the app will display the DataFrame with a downloadable CSV option which includes the status column as shown below.

Sample Output

Please note that if there is any error encountered regarding closing the connection, you should cache the connection using the @st.cache_resource decorator on the connection functions.

Thanks for Reading !!