如何使用Python截断Snowflake中的表？

Question

我编写了一个简单的 python 函数，它将检查 Snowflake 中是否存在表，如果存在则将截断它：

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))

它检查表是否正确存在，然后继续运行截断。但我收到一条警告消息：

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.

然后当我检查 Snowflake 时，这张表没有被截断。

该功能的使用方法如下：

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()

在 Snowflake 工作表中，我可以运行大写或小写表名的截断表，没有任何问题。

有人可以告诉我为什么它不截断并在 python 中给出此消息吗？

Answer 1

我建议在 SQL 查询中使用 Snowflake 的参数绑定而不是 Python 的 f 字符串格式。您可以在 Snowflake 文档中找到示例。

以下是修改代码的方法：

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})

我希望这有帮助！让我知道它是否适合您！