如何在Snowflake Python连接器中通过Query ID使用时间旅行？

问题场景 #

使用Snowflake变量的方式可以正常实现时间旅行查询：

snowflake_cursor.execute("""
set q_id = last_query_id();""")
snowflake_cursor.execute(""" 
select max(scenario_id) from scenarios at(statement=>$q_id);""")

但Snowflake变量仅在当前连接会话内有效，若想将Query ID存入Python变量，以便后续重新连接后仍能使用时间旅行，尝试以下代码时会触发unexpected '$'错误：

snowflake_cursor.execute("""
select last_query_id();
""")
q_id = snowflake_cursor.fetchone()[0]
snowflake_cursor.execute("""
select max(scenario_id) from scenarios at(statement=>$%s);
""", q_id)

解决方法 #

错误原因是混淆了Snowflake会话变量的$前缀与Python连接器的参数化占位符语法。以下是两种可行方案：

方案1：使用参数化查询（推荐，避免SQL注入）

直接用Python连接器的%s占位符传递Query ID，无需添加$前缀：

snowflake_cursor.execute("select last_query_id();")
q_id = snowflake_cursor.fetchone()[0]
# 正确的参数化写法，将q_id作为参数传入
snowflake_cursor.execute("select max(scenario_id) from scenarios at(statement=>%s);", (q_id,))

这种方式由连接器自动处理字符串转义，安全性更高，适合大多数场景。

方案2：字符串格式化（仅适用于Query ID为可信来源的场景）

如果能确保Query ID不会包含恶意内容，也可以直接用Python的字符串格式化拼接SQL语句：

snowflake_cursor.execute("select last_query_id();")
q_id = snowflake_cursor.fetchone()[0]
# 用f-string格式化SQL，注意单引号包裹Query ID
snowflake_cursor.execute(f"select max(scenario_id) from scenarios at(statement=>'{q_id}');")

内容的提问来源于stack exchange，提问作者slothish1