snowflake-core

import os from snowflake.connector import connect from snowflake.snowpark import Session from snowflake.core import Root # Option 1: Create Root instance from Snowflake Connection connection_params = { "account": os.environ.get("SNOWFLAKE_ACCOUNT", ""), "user": os.environ.get("SNOWFLAKE_USER", ""), "password": os.environ.get("SNOWFLAKE_PASSWORD", ""), "database": os.environ.get("SNOWFLAKE_DATABASE", "test_db"), "warehouse": os.environ.get("SNOWFLAKE_WAREHOUSE", "test_wh"), "schema": os.environ.get("SNOWFLAKE_SCHEMA", "public"), } # Ensure required environment variables are set if not all(connection_params.values()): print("Warning: Some Snowflake connection parameters are missing. Please set SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, and SNOWFLAKE_PASSWORD environment variables.") # Exit or handle error if connection is critical try: # Using snowflake.connector connection = connect(**connection_params) root_from_conn = Root(connection) print(f"Root instance created from SnowflakeConnection: {root_from_conn}") # Example: Accessing a database collection # Assuming 'mydatabase' exists or will be created # my_database = root_from_conn.databases["MYDATABASE"] # print(f"Accessed database: {my_database}") connection.close() # Option 2: Create Root instance from Snowpark Session # Ensure required environment variables are set for Snowpark as well if different session = Session.builder.configs(connection_params).create() root_from_session = Root(session) print(f"Root instance created from Snowpark Session: {root_from_session}") # Example: Accessing a compute pool collection # compute_pools = root_from_session.compute_pools # print(f"Accessed compute pools collection: {compute_pools}") session.close() except Exception as e: print(f"An error occurred during quickstart: {e}")