snowflake_connector

SnowflakeConnector

Connector class for Snowflake operations related to Slack interactions. Handles connecting to Snowflake and provides methods for data operations.

source

SnowflakeConnector

 SnowflakeConnector (connection_params:Optional[Dict[str,Any]]=None)

Connector class for Snowflake operations related to Slack interactions. Handles connecting to Snowflake and provides methods for data operations.

First, if we want to make a Snowfalke Connection class, we should probably make a way to connect to Snowflake…

Yay! We are now able to connect to Snowflake which is pretty cool. Let’s also make a method to close the connection when we are done with it. This is important because we don’t want to leave connections open and use up resources. We can do this by adding a close method to our class. This method will call the close method on the connection object. We can also add a check to see if the connection is already closed before trying to close it again.

source

SnowflakeConnector.close

 SnowflakeConnector.close ()

Close the Snowflake connection if open.

Awesome, now we can connect to Snowflake and close the connection when we are done with it. Let’s also add a method to execute a query and return the results. This will be useful for getting data from Snowflake. We can do this by adding an execute_query method to our class. This method will take a query string as an argument and return the results of the query. We can use the cursor object to execute the query and fetch the results.

source

SnowflakeConnector.execute_query

 SnowflakeConnector.execute_query (query:str,
                                   params:Optional[List[Any]]=None)

*Execute a query and return results as a list of dictionaries.

Args: query: SQL query to execute params: Optional parameters for the query

Returns: List of dictionaries with query results*

Data Insertion

Let’s also create a way to insert data into Snowflake.

source

SnowflakeConnector.insert_record

 SnowflakeConnector.insert_record (table_name:str, data:Dict[str,Any],
                                   **kwargs)

*Enhanced function to insert data into Snowflake with improved type handling.

Args: table_name: Name of the table to insert into data: Dictionary of data to insert **kwargs: Additional options for insertion

Returns: ID of the inserted record (if id_field is provided) or True if successful*

Now that we can insert individual rows, let’s also add a method to insert multiple rows at once

source

SnowflakeConnector.bulk_insert

 SnowflakeConnector.bulk_insert (table_name:str,
                                 df:pandas.core.frame.DataFrame, **kwargs)

*Enhanced function to bulk insert DataFrame data into Snowflake with improved type handling.

Args: table_name: Name of the table to insert into df: DataFrame with columns matching table structure **kwargs: Additional options for insertion - database: Override default database - schema: Override default schema - auto_timestamp: Add timestamps to standard timestamp fields (default: True) - timezone: Timezone for timestamps (default: ‘America/Chicago’) - debug: Enable debug output (default: False) - batch_size: Rows per batch (default: 10000) - quote_identifiers: Whether to quote identifiers (default: False) - chunk_size: Size of DataFrame chunks to process (default: None) - auto_create_table: Create table if not exists (default: False) - use_column_mapping: Map DataFrame columns to table columns (default: True)

Returns: True if successful*

source

SnowflakeConnector.__del__

 SnowflakeConnector.__del__ ()

Ensure connection is closed when object is destroyed.

source

SnowflakeConnector.get_interaction_summary

 SnowflakeConnector.get_interaction_summary ()

*Get a summary of interactions by view and action type.

Returns: Summary statistics*

source

SnowflakeConnector.get_interactions_by_view

 SnowflakeConnector.get_interactions_by_view (view:str, limit:int=100)

*Get recent interactions for a specific view.

Args: view: View name limit: Maximum number of records to return

Returns: List of interaction records*

source

SnowflakeConnector.get_user_interactions

 SnowflakeConnector.get_user_interactions (user_id:str, limit:int=100)

*Get recent interactions for a specific user.

Args: user_id: Slack user ID limit: Maximum number of records to return

Returns: List of interaction records*