如何在Python项目中验证存在引号不平衡问题的SQL查询

需求明确 #

需要在Python项目里解析MySQL查询，揪出以下几类引号相关的错误查询：

• 引号不平衡（比如字符串未闭合）
• 引号位置错误（字符串未收尾就续写其他条件）
• 列过滤值引号使用不当（把条件逻辑符号写到字符串里）

典型错误示例：

1. select * from city where name='x and type=y; —— 错误（缺少闭合单引号）
2. select * from city where name='x and type=y'; —— 错误（字符串'x未闭合，后续条件被误判为字符串内容）

已尝试的工具与代码 #

试过sqlparse、sqlglot、sqlvalidator三种工具，其中用sqlglot的尝试代码如下：

from sqlglot import exp, parse_one
try:
     sqlglot.transpile("select * from city where name='x and type=y';")
except sqlglot.errors.ParseError as e:
     print(e.errors)

优化后的检测方案（基于sqlglot） #

sqlglot的语法解析能力能精准捕获这类引号相关的语法错误，下面是更实用的封装函数：

import sqlglot
from sqlglot.errors import ParseError

def check_mysql_quote_errors(sql_query):
    """检测MySQL查询中的引号类语法错误"""
    try:
        # 指定MySQL方言解析，适配MySQL语法规则
        sqlglot.parse_one(sql_query, read="mysql")
        return None  # 无错误返回None
    except ParseError as e:
        # 筛选出和引号、未闭合字符串相关的错误信息
        target_errors = []
        for err_msg in e.errors:
            err_lower = err_msg.lower()
            if "quote" in err_lower or "unclosed" in err_lower or "string" in err_lower:
                target_errors.append(err_msg)
        return target_errors if target_errors else str(e)

# 测试示例查询
test_queries = [
    "select * from city where name='x and type=y;",
    "select * from city where name='x and type=y';"
]

for idx, sql in enumerate(test_queries, 1):
    errors = check_mysql_quote_errors(sql)
    if errors:
        print(f"示例{idx}错误：{errors}")
    else:
        print(f"示例{idx}查询合法")

这个方案的优势：

• 针对MySQL方言解析，兼容性更强
• 自动过滤出引号相关的错误，避免无关报错干扰
• 支持批量检测多段查询

补充：用sqlparse做快速引号检测 #

如果只需要快速检测引号闭合情况，sqlparse的token级检查更轻量：

import sqlparse
from sqlparse.tokens import Token

def check_quotes_with_sqlparse(sql_query):
    """用sqlparse检测引号闭合与字符串内的异常符号"""
    tokens = sqlparse.parse(sql_query)[0].flatten()
    quote_stack = []
    # 检查引号闭合情况
    for token in tokens:
        if token.ttype in (Token.String.Single, Token.String.Double):
            if quote_stack and quote_stack[-1] == token.ttype:
                quote_stack.pop()
            else:
                quote_stack.append(token.ttype)
    if quote_stack:
        return "存在未闭合的引号"
    # 简单检测字符串内是否混入条件符号
    for token in tokens:
        if token.ttype == Token.String.Single and '=' in token.value:
            return "字符串值中可能包含错误的条件符号（引号使用不当）"
    return None

# 测试
for idx, sql in enumerate(test_queries, 1):
    errors = check_quotes_with_sqlparse(sql)
    if errors:
        print(f"示例{idx}错误：{errors}")

注意：sqlparse的检测更偏向于token层面，对复杂语法逻辑错误的识别需要自定义补充规则。

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