Python向Snowflake插入含单引号JSON数据时报错的解决方法

使用Python脚本将REST API获取的字典数据插入Snowflake的VARIANT类型列时，因JSON内容包含单引号触发SQL语法错误，更换为原生snowflake.connector库操作仍会复现该问题。
原有问题代码如下：

from sqlalchemy import create_engine
import urllib
import requests
import json


engine = create_engine(
'snowflake://{user}:{password}@{account_identifier}/'.format(
    user='UserName',
    password='pwd',
    account_identifier='account_Identifier'
 )
)

jval={"title":"value", "address":[{ "Road":"st xavier's","landmark":"D' Pauls"}]}


try:
    connection = engine.connect()
    results = connection.execute(
        "INSERT INTO db_name.schema_name.sqlalchemy(jval) (select PARSE_JSON('" + json.dumps(jval) + "'))").fetchone()
    print(results[0])
finally:
    connection.close()
    engine.dispose()

触发的报错信息：

sqlalchemy.exc.ProgrammingError: (snowflake.connector.errors.ProgrammingError) 001003 (42000): SQL compilation error:
syntax error line 1 at position 133 unexpected 's'.
syntax error line 1 at position 134 unexpected '", "'.
[SQL: INSERT INTO db_name.schema_name.sqlalchemy(jval)  (select PARSE_JSON('{"title": "value", "address": [{"Road": "st xavier's", "landmark": "D' Pauls"}]}'))]

问题出在手动拼接SQL字符串的逻辑：Snowflake中字符串常量使用单引号包裹，序列化后的JSON内容本身包含单引号（比如示例里的st xavier's、D' Pauls）时，这些单引号会被SQL解析器识别为字符串的结束边界，直接截断后续内容，最终触发语法错误。

核心原则：永远不要手动拼接SQL传入参数，使用数据库驱动提供的参数绑定（参数化查询）能力，驱动会自动完成所有特殊字符的转义，同时避免SQL注入风险。

推荐写法（SQLAlchemy版本） #

使用%s作为参数占位符，将序列化后的JSON字符串作为参数传入execute方法，无需手动处理任何转义：

from sqlalchemy import create_engine
import json

engine = create_engine(
'snowflake://{user}:{password}@{account_identifier}/'.format(
    user='UserName',
    password='pwd',
    account_identifier='account_Identifier'
 )
)

jval = {"title":"value", "address":[{ "Road":"st xavier's","landmark":"D' Pauls"}]}
json_str = json.dumps(jval)

try:
    connection = engine.connect()
    # 参数占位符为%s，参数通过元组传入execute方法
    insert_sql = "INSERT INTO db_name.schema_name.sqlalchemy(jval) SELECT PARSE_JSON(%s)"
    results = connection.execute(insert_sql, (json_str,)).fetchone()
    print(results[0])
finally:
    connection.close()
    engine.dispose()

注：高版本的Snowflake驱动支持直接传入Python字典对象作为VARIANT类型参数，无需手动调用json.dumps()，驱动会自动完成类型转换。

原生snowflake.connector写法 #

逻辑完全一致，同样使用参数绑定即可：

import snowflake.connector
import json

conn = snowflake.connector.connect(
    user='UserName',
    password='pwd',
    account='account_Identifier',
    warehouse='YOUR_WH',
    database='YOUR_DB',
    schema='YOUR_SCHEMA'
)

jval = {"title":"value", "address":[{ "Road":"st xavier's","landmark":"D' Pauls"}]}
json_str = json.dumps(jval)

try:
    cur = conn.cursor()
    insert_sql = "INSERT INTO sqlalchemy(jval) SELECT PARSE_JSON(%s)"
    cur.execute(insert_sql, (json_str,))
    print(cur.fetchone()[0])
finally:
    cur.close()
    conn.close()

不推荐的临时方案 #

如果场景特殊必须拼接SQL，需要手动将JSON字符串内的所有单引号替换为两个单引号（Snowflake SQL中单个单引号的转义格式为''）：

# 仅作演示，生产环境禁止使用
escaped_json = json.dumps(jval).replace("'", "''")
insert_sql = f"INSERT INTO db_name.schema_name.sqlalchemy(jval) SELECT PARSE_JSON('{escaped_json}')"

该方式存在SQL注入风险，且遇到其他特殊字符时仍可能触发错误，不建议使用。

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