How can I truncate table in Snowflake using Python?

Question

I have written a simple Python function that will check to see if a table exists in Snowflake, and if so will truncate it:

def truncate_if_exists(self, connection_session, table_name):
    """Truncates table if it exists"""

    check_query = f"""
        SELECT 1
        FROM INFORMATION_SCHEMA.TABLES
        WHERE TABLE_NAME = '{table_name}'
    """
    truncate_query = f"""
        TRUNCATE TABLE {table_name}
    """
    exists = connection_session.execute(text(check_query)).scalar()
    if exists:
        connection_session.execute(text(truncate_query))

It checks to see that table exists correctly, proceeds to run the truncation. But I get a warning message:

UserWarning: The provided table name 'TEST_TABLE' is not found exactly as such in the database after writing the table, possibly due to case sensitivity issues. Consider using lower case table names.

Then when I check in Snowflake, this table did not get truncated.

Here's how the function is used:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import text

conn_string = 'snowflake://{username}:{password}@{account_identifier}/{schema}/{database}?warehouse={warehouse}&role=SYSADMIN'
conn_engine = create_engine(conn_string)
conn_session = sessionmaker(bind=conn_engine)

land_table = 'TEST_TABLE'
session = conn_session()    
truncate_if_exists(session, land_table)
session.close()

In the Snowflake worksheet, I can run truncate table with upper or lower case table name no problems.

Why is it not truncating and giving this message in Python?

Answer 1

I suggest using Snowflake's parameter binding instead of Python's f-string formatting for your SQL queries. Examples are in the Snowflake documentation.

Here's how you could modify your code:

def truncate_if_exists(self, connection_session, table_name):
    """Truncates table if it exists"""

    check_query = """
        SELECT 1
        FROM INFORMATION_SCHEMA.TABLES
        WHERE TABLE_NAME = %(table_name)s
    """
    truncate_query = """
        TRUNCATE TABLE %(table_name)s
    """
    exists = connection_session.execute(check_query, {"table_name": table_name}).scalar()
    if exists:
        connection_session.execute(truncate_query, {"table_name": table_name})