一直做后端开发绕不开跟 SQL 打交道。但你有没有认真想过一个问题拿到一串 SQL 文本纯靠字符串处理去拆分、格式化、提取信息这条路到底能走多远我早期接过一个蛮折腾的活儿把上百个 SQL 脚本做批量规范化统一关键字大小写、统一缩进还要自动统计每条语句操作了哪些表。一开始图省事用正则结果一遇到多层子查询、字符串里藏分号、注释块正则直接原地爆炸。后来换成 Python 的 sqlparse 库原本要熬一个通宵的活一下午就给整明白了。sqlparse 是 Python 生态里一个非常低调但实用的 SQL 解析库零第三方依赖安装后就能用。它的定位不是像数据库内核那样生成完整的语法树而是提供一套实用的语句切割、格式化、Token 级识别能力。如果你做 Python 后端、搞数据分析、处理慢 SQL 日志或者想快速做出一个 SQL 静态检查小工具这个库几乎是最轻量顺手的那一个。这篇文章我会把它最核心的四个能力split、parse、format、tokenize拆开一条条讲结合真实场景给出可直接复制的代码再把我实际踩过的一些坑也一并交代清楚。1. SQL文本处理为什么需要专门的解析库1.1 正则处理SQL的天然短板很多人一开始都像我一样想用正则把 SQL 给收拾了。单独看几条“干净”的 SQL正则确实能应付。但现实世界里的 SQL 远没有那么听话。比如下面这条select count(*) from orders where note its a semi;colon test; -- comment;如果你简单用分号去 split(\u003b)字符串里那个分号会直接把语句切漏。再比如注释里出现的关键字、数字或者一行 SQL 里带了/* ... */多行注释正则都需要配套考虑。更别提像WITH ... AS、子查询嵌套、CASE WHEN 这种结构正则想靠几条 pattern 就能完整覆盖各种组合基本是给自己挖坑。一个比较贴切的比喻正则是用一把大剪刀去拆电路板上的元件稍不注意就会剪断旁边的绝缘层破坏整块板子。那为什么不能干脆写个完整的 SQL 解析器因为工作量和维护成本太高。SQL 的语法是按数据库方言走的MySQL、PostgreSQL、Oracle、SQL Server 各自的细节差异很大想支持全部细节都已经不是个人项目而是产品级团队的活。sqlparse 的思路很聪明它不追求“完全理解语义”而是做到“可靠地切词、分组和格式化”把最脏最累的结构化文本处理活了干漂亮至于高层次的语义判断留给上层应用去决定。1.2 sqlparse的定位和实现思路sqlparse 是一个纯 Python 实现的库核心逻辑是词法分析和基础语法分组。它先用类似状态机的方式把 SQL 文本切成 Token再把这些 Token 组合成 Statement、Identifier、Where、Parenthesis 等有结构的分组对象。这意味着它的好处很直接不需要装任何编译工具链pip install就能跑。基于 Python 自带的正则引擎跨平台能力好。Token 和分组类型设计得相对完整足够做很多实用的二次加工。它的局限也很直接它给出的不是一棵完整抽象语法树也不会帮你解析每种数据库方言的专属语法。比如 SQL Server 特有的TOP、Oracle 的CONNECT BY在分 Token 层面不会出大问题但如果想准确把握语义还得在分组基础上自己再加工。1.3 sqlparse与专业解析器的界限市面上还有其他选择我简单列个对比方便你按场景选型工具定位适合场景sqlparse轻量 SQL 文本处理快速拆分、格式化、基础信息提取sqlglot跨方言 SQL 解析与转译需要方言翻译、AST 级改写sqlfluffSQL Lint 与规范检查团队 SQL 风格统一、代码质量门禁ANTLR 自定义语法完整语法解析自研数据库协议、复杂 AST 需求如果你只是想把一堆 SQL 从“乱糟糟的字符串”变成“干净的、可继续加工的数据”sqlparse 是最快的一条路。它像是瑞士军刀小巧但够用。真要到做数据库协议解析那一步再考虑更重的方案也不会亏。2. 核心能力速览与API走读2.1 环境准备与安装安装只需要一条命令pip install sqlparse装完之后可以顺手确认下版本import sqlparse print(sqlparse.__version__)sqlparse 官方给出的定位是“无依赖、纯 Python”也就是你的虚拟环境里不会多出一堆连锁依赖。这点对开发调试非常友好尤其当你需要在生产环境快速部署一个临时脚本时体量小能省很多事。2.2 split把大脚本拆成单条语句sqlparse.split的逻辑很简单就是把一大段文本按语句边界切开返回一个字符串列表。它的智能之处在于能正确处理分号出现在字符串常量、注释、存储过程等场景中时不会被误切。import sqlparse sql select id, name from users where status 1; update users set status 0 where id 1; /* this is a comment; do not split here */ select count(*) from logs; statements sqlparse.split(sql) for i, s in enumerate(statements, 1): print(f语句{i}: {s[:50]}...)输出效果大致是语句1: select id, name from users where status 1; 语句2: update users set status 0 where id 1; 语句3: /* this is a comment; do not split here */ 语句4: select count(*) from logs;我实际用下来的感受是它对于大多数标准 SQL、存储过程脚本的分割都比较可靠这是手工正则很难做到的。2.3 parse返回Statement对象列表parse返回的是Statement对象列表每个Statement保留了结构化的 Token 分组信息。和split的区别在于parse不只是拿到纯文本片段而是能继续往下做结构化访问。parsed sqlparse.parse(sql) for stmt in parsed: print(stmt.get_type())get_type()会返回SELECT、UPDATE、CREATE这类首关键字非常适合拿来做语句类型分流。比如把输入的 SQL 分成 DDL、DML 两类再决定是否允许执行这在工单系统里很常用。还可以直接遍历Statement的tokens属性for stmt in parsed: for token in stmt.tokens: print(token.ttype, repr(token.value))这里会发现一个 Statement 是一个“容器”它内部有DML如SELECT、Keyword、Name、Whitespace、Punctuation等各种各样的 Token。掌握这个视角之后很多任务就变得像“在抽屉里按标签取东西”一样简单。2.4 format格式化SQL文本format是我日常用得最多的功能。它可以把一团乱 SQL 整理成风格统一、可读性高的样子。常用的参数有参数作用我的推荐值keyword_case关键字大小写upper或loweridentifier_case标识符大小写lower很多团队用strip_comments是否删除注释按需开启reindent是否重新缩进Truereindent_aligned对齐方式是否按列对齐False否则 diff 太大use_space_around_operators运算符两侧加空格Truecomma_first逗号是否前置看团队风格示例raw_sql select id,name,price from products where price100 order by id desc formatted sqlparse.format( raw_sql, keyword_caseupper, identifier_caselower, reindentTrue, use_space_around_operatorsTrue, ) print(formatted)输出SELECT id, name, price FROM products WHERE price 100 ORDER BY id DESC这种格式化在慢 SQL 分析、输出到日志、生成报表场景里非常顺手。你不要小看这几行代码它能省下大量肉眼比对 SQL 的时间。2.5 tokenize接口与Token结构如果你想从底层理解一条 SQL应该直接接触Token结构。sqlparse.tokenize会把文本切成一串 Token不按语句做进一步的归组适合做深度文本分析。from sqlparse import tokenize from sqlparse.tokens import Keyword, DML, String, Comment sql select * from users -- a comment for token in tokenize(sql): print(token.ttype, repr(token.value))常见 Token 类别包括Keywordselect、from、where、join等。DMLINSERT、UPDATE、DELETE这类数据操作关键字。Name表名、列名等标识符。String单引号或双引号包裹的字符串。Number数字。Comment注释包括--和/* */。还记得我们最初提到的正则痛点吗一旦有了 Token 类型按类型过滤就变得特别省心。比如想“去掉 SQL 里所有注释”from sqlparse import parse stmt parse(select * from users -- comment test)[0] clean_value .join( token.value for token in stmt.flatten() if token.ttype not in (Comment,) ) print(clean_value)flatten()会递归展开嵌套的分组 Token这个技巧在写各类 SQL 清洗工具时特别实用。3. 格式化与规范化实战3.1 快速实现“常量化”的SQL指纹工具做慢 SQL 优化时最头疼的事情之一是“同一条 SQL 只是条件值不同却被当成几百条记录”。这时候如果能把它们归并成一个指纹就方便统计 top N 相似 SQL。我的思路是先把 SQL 标准化用strip_comments去注释把字符串和数字替换成统一的占位符再调用format统一关键字大小写。这样同一类 SQL 的不同参数值就不会影响后续分组统计了。import sqlparse from sqlparse.tokens import String, Number def sql_fingerprint(sql_text): parsed sqlparse.parse(sqlparse.format( sql_text, strip_commentsTrue, keyword_caselower ))[0] tokens [] for token in parsed.flatten(): if token.ttype in (String, Number): tokens.append(?) else: tokens.append(token.value) normalized .join(tokens) normalized sqlparse.format( normalized, keyword_caselower, reindentTrue ) return normalized这个方案的核心是把“语义等价但字面量不同”的 SQL 映射到同一个文本串。实际跑在慢查询日志上原本看起来五花八门的 SQL可能最后只收敛到十几条核心模板优化方向立刻清晰很多。3.2 给慢SQL日志做标准化处理慢 SQL 日志往往是几百兆甚至上 GB 的大文件里面除了 SQL 本身还有时间、耗时、锁等待时间等元数据。用脚本逐行读出来再交给 sqlparse 去拆分和格式化是我处理这一类问题的标准姿势。import re import sqlparse slow_log_pattern re.compile( r# Query_time:\s([\d.])\s.*? r#\s r(.*?)(?# Query_time:|\Z), re.DOTALL ) def parse_slow_log(filepath): with open(filepath, r, encodingutf-8, errorsignore) as f: content f.read() for match in slow_log_pattern.finditer(content): query_time float(match.group(1)) sql_block match.group(2).strip() for statement in sqlparse.split(sql_block): formatted sqlparse.format( statement, keyword_caseupper, strip_commentsTrue, reindentTrue ) yield query_time, formatted这段逻辑其实是想说明一个通用步骤先用元信息把有效 SQL 段抽出来再用 sqlparse 拆成单条、格式化统一风格。格式化后的 SQL 方便写入中间表或者传给分析工具做聚合后续定位慢查询根因会轻松太多。3.3 借助pre-commit统一团队SQL风格我去年给团队做过一个很实用的东西在 git 的 pre-commit 阶段跑一个 py 脚本检查提交的.sql文件里的关键字大小写和基本缩进格式不符合规则就直接报错。实现思路就是用format后的文本和原文本比对。import sys import sqlparse def lint_sql_file(filepath): with open(filepath, r, encodingutf-8) as f: raw f.read() formatted sqlparse.format( raw, keyword_caseupper, identifier_caselower, strip_commentsFalse, reindentTrue, use_space_around_operatorsTrue, ) if formatted ! raw: print(fSQL风格不合格: {filepath}) return False return True if __name__ __main__: fail False for path in sys.argv[1:]: if path.endswith(.sql): fail not lint_sql_file(path) or fail sys.exit(1 if fail else 0)这里有个坑不要直接在钩子里自动改写文件否则会频繁制造大 diff队友容易心态爆炸。改成报告错误、让提交者手动修改效果反而更好。4. 解析能力用于安全检查与信息提取4.1 基于Statement类型分流在内部工单系统里经常需要区分用户提交的 SQL 是查询类还是变更类。用get_type()就能很快实现import sqlparse def classify_sql(sql_text): results [] for stmt in sqlparse.parse(sql_text): sql_type stmt.get_type().upper() if sql_type in (SELECT, SHOW, EXPLAIN): results.append((read, stmt)) elif sql_type in (INSERT, UPDATE, DELETE, MERGE, REPLACE): results.append((write, stmt)) elif sql_type in (CREATE, ALTER, DROP, TRUNCATE): results.append((ddl, stmt)) else: results.append((other, stmt)) return results有了这个分类可以在审批、审计、风险提示上做很多文章。比如 DDL 自动触发负责人审批SELECT类直接放行DELETE类强制二次确认。这不是危险功能而是把流程自动化减少人为疏忽。4.2 从SQL中提取表名提取表名看起来容易实际上一堆 SQL 风格差异很大有人写FROM users有人写FROM db.users u还有人写FROM (select ...) t。完整的表名抽取需要处理各种情况但 sqlparse 给了很好的起点。我用过的一个相对稳妥的简化策略import sqlparse from sqlparse.sql import Identifier, IdentifierList from sqlparse.tokens import Keyword, DML def extract_tables(sql_text): tables [] parsed sqlparse.parse(sql_text) for stmt in parsed: for token in stmt.tokens: if token.ttype in (DML, Keyword) and token.value.upper() in (FROM, JOIN, UPDATE, INTO): next_token token # 这里要拿到 FROM 后面的部分需要借助 sqlparse 对语句结构的识别 if isinstance(token, Identifier): tables.append(token.get_real_name()) elif isinstance(token, IdentifierList): for ident in token.get_identifiers(): tables.append(ident.get_real_name()) return tables实际上FROM后面的内容可能是Identifier、IdentifierList或一个括号子查询需要边界判断。更成熟的实现可以遍历 token 列表记录关键字的索引再取后续的标识符。这个函数不像官方库那样保证 100% 正确但结合正则和人工审查在文档生成、血缘分析场景已经能发挥大作用。4.3 辅助SQL注入风险检测聊到注入我只聊防御方向的思路。把用户输入的 SQL 绑定成参数化查询永远是第一位但很多遗留系统里确实还有拼接 SQL 的情况。写一个轻量检测工具把可能有风险的 SQL 挑出来告警是可行的。import sqlparse from sqlparse.tokens import Comment def has_risk(sql_text): parsed sqlparse.parse(sql_text) if not parsed: return False stmt parsed[0] # 检查注释很多绕过手法会在注释上做文章 for token in stmt.flatten(): if token.ttype is Comment: return True # 检测关键字组合 keywords [token.value.upper() for token in stmt.flatten() if token.ttype in sqlparse.tokens.Keyword] if UNION in keywords and SELECT in keywords: return True # 检测恒真条件如 OR 11 sql_upper stmt.value.upper() if OR 11 in sql_upper or OR 1 1 in sql_upper: return True return False这段代码只能覆盖很浅的特征千万不要把它当成万无一失的安全防线。真正安全还是要靠参数绑定、最小权限、SQL 白名单和源头规范化。sqlparse 在这里的角色是辅助过滤帮人早一步发现问题而不是替代防护体系。4.4 批量抽取DDL生成数据字典有一次我需要给几十张表自动整理成 Markdown 数据字典每张表结构都写一遍太耽误时间。我写了一个小脚本用 sqlparse 解析CREATE TABLE语句提取表名和字段清单再拼成表格输出。import sqlparse from sqlparse.tokens import Name, Punctuation def extract_columns(create_stmt): columns [] # 只解析 CREATE TABLE 语句的括号内部 for token in create_stmt.tokens: if isinstance(token, sqlparse.sql.Parenthesis): for subtoken in token.tokens: if subtoken.ttype is Name: columns.append(subtoken.value) return columns for stmt in sqlparse.parse(open(schema.sql).read()): if stmt.get_type() CREATE: table_name None # 通过 token 分析拿到表名 for token in stmt.tokens: if token.ttype is Name: table_name token.value break print(f表: {table_name}) print(extract_columns(stmt))这类脚本的价值不在技术难度而在于节省重复劳动。遇到成百上千张表时手写整理文档绝对是一场灾难。5. 踩坑清单与性能实测5.1 我踩过的4个坑第一split和parse的边界行为不完全一样。如果一段文本里有分号前后没有内容split()可能会过滤掉空串但parse()可能得到一个空的 Statement。写统计脚本时要注意空对象依然存在直接调get_type()可能返回空值或异常。第二format(reindentTrue)和format(reindent_alignedTrue)混用时输出可能和你预期的不一致。reindent_aligned想把对齐做得很细但自动生成的格式不一定符合团队习惯而且提交到 git 上会产生比较大的改动。建议先选定一种模式不要两个参数同时开。第三对特别复杂的大存储过程做格式化耗时会明显上升。sqlparse 虽然轻量但不是免费的午餐。它会将整条 SQL 放到内存里构建结构对于几十 MB 的超级脚本建议先split成单条再逐条处理避免一次解析过长文本导致内存抖动。第四get_type()不是万能的。遇到语法不完整的片段比如中途截断的 SQLget_type()可能返回UNKNOWN。不要盲目信任处理用户输入时要做好防御。5.2 性能和内存表现我用一台普通开发机8 核 16G处理过大约 10 万条常见 SQL 片段。每条平均几个关键字纯parse耗时总体在 10 秒量级如果只做format会更快一点。单条 SQL 如果特别长比如上千行的存储过程耗时可能在几十到几百毫秒。这类场景非常适合用多进程并行Python 的concurrent.futures.ProcessPoolExecutor就能直接上import concurrent.futures def process_one(sql_text): return sqlparse.format(sql_text, keyword_caseupper, reindentTrue) sql_list [...] # 从日志中切分好的 SQL with concurrent.futures.ProcessPoolExecutor(max_workers8) as executor: results list(executor.map(process_one, sql_list))实测中多进程能明显提升吞吐。注意不要在子进程里反复 import 过大量初始化的全局资源保持任务尽量纯函数化。5.3 命令行快速格式化工具最后分享一个能长期提升体验的小技巧把 sqlparse 封装成命令行工具随时粘贴 SQL 或者从文件输入格式化后的结果直接输出。cat messy.sql | python -c import sys, sqlparse sql sys.stdin.read() print(sqlparse.format(sql, keyword_caseupper, reindentTrue, use_space_around_operatorsTrue)) 你也可以把它写成一个.py脚本放到自己的工具目录里以后分析、分享 SQL 都方便很多。我在内部还把它做成了一个小小的缓存命令队友给我贴 SQL我格式化完再看效率高不少。sqlparse 这个库最大的价值是让你不必在正则和字符串拼接的泥潭里挣扎就能把 SQL 当结构化对象去处理。从批量格式化、指纹归并到分类提取、辅助风险识别它都是背后那个踏实好用的“隐形工具人”。如果你正好也想处理一堆 SQL 文本建议先拿split和format练手很快你会发现原来被字符串折磨的日子也可以这么清爽。
sqlparse：Python轻量级SQL解析与格式化实战指南
一直做后端开发绕不开跟 SQL 打交道。但你有没有认真想过一个问题拿到一串 SQL 文本纯靠字符串处理去拆分、格式化、提取信息这条路到底能走多远我早期接过一个蛮折腾的活儿把上百个 SQL 脚本做批量规范化统一关键字大小写、统一缩进还要自动统计每条语句操作了哪些表。一开始图省事用正则结果一遇到多层子查询、字符串里藏分号、注释块正则直接原地爆炸。后来换成 Python 的 sqlparse 库原本要熬一个通宵的活一下午就给整明白了。sqlparse 是 Python 生态里一个非常低调但实用的 SQL 解析库零第三方依赖安装后就能用。它的定位不是像数据库内核那样生成完整的语法树而是提供一套实用的语句切割、格式化、Token 级识别能力。如果你做 Python 后端、搞数据分析、处理慢 SQL 日志或者想快速做出一个 SQL 静态检查小工具这个库几乎是最轻量顺手的那一个。这篇文章我会把它最核心的四个能力split、parse、format、tokenize拆开一条条讲结合真实场景给出可直接复制的代码再把我实际踩过的一些坑也一并交代清楚。1. SQL文本处理为什么需要专门的解析库1.1 正则处理SQL的天然短板很多人一开始都像我一样想用正则把 SQL 给收拾了。单独看几条“干净”的 SQL正则确实能应付。但现实世界里的 SQL 远没有那么听话。比如下面这条select count(*) from orders where note its a semi;colon test; -- comment;如果你简单用分号去 split(\u003b)字符串里那个分号会直接把语句切漏。再比如注释里出现的关键字、数字或者一行 SQL 里带了/* ... */多行注释正则都需要配套考虑。更别提像WITH ... AS、子查询嵌套、CASE WHEN 这种结构正则想靠几条 pattern 就能完整覆盖各种组合基本是给自己挖坑。一个比较贴切的比喻正则是用一把大剪刀去拆电路板上的元件稍不注意就会剪断旁边的绝缘层破坏整块板子。那为什么不能干脆写个完整的 SQL 解析器因为工作量和维护成本太高。SQL 的语法是按数据库方言走的MySQL、PostgreSQL、Oracle、SQL Server 各自的细节差异很大想支持全部细节都已经不是个人项目而是产品级团队的活。sqlparse 的思路很聪明它不追求“完全理解语义”而是做到“可靠地切词、分组和格式化”把最脏最累的结构化文本处理活了干漂亮至于高层次的语义判断留给上层应用去决定。1.2 sqlparse的定位和实现思路sqlparse 是一个纯 Python 实现的库核心逻辑是词法分析和基础语法分组。它先用类似状态机的方式把 SQL 文本切成 Token再把这些 Token 组合成 Statement、Identifier、Where、Parenthesis 等有结构的分组对象。这意味着它的好处很直接不需要装任何编译工具链pip install就能跑。基于 Python 自带的正则引擎跨平台能力好。Token 和分组类型设计得相对完整足够做很多实用的二次加工。它的局限也很直接它给出的不是一棵完整抽象语法树也不会帮你解析每种数据库方言的专属语法。比如 SQL Server 特有的TOP、Oracle 的CONNECT BY在分 Token 层面不会出大问题但如果想准确把握语义还得在分组基础上自己再加工。1.3 sqlparse与专业解析器的界限市面上还有其他选择我简单列个对比方便你按场景选型工具定位适合场景sqlparse轻量 SQL 文本处理快速拆分、格式化、基础信息提取sqlglot跨方言 SQL 解析与转译需要方言翻译、AST 级改写sqlfluffSQL Lint 与规范检查团队 SQL 风格统一、代码质量门禁ANTLR 自定义语法完整语法解析自研数据库协议、复杂 AST 需求如果你只是想把一堆 SQL 从“乱糟糟的字符串”变成“干净的、可继续加工的数据”sqlparse 是最快的一条路。它像是瑞士军刀小巧但够用。真要到做数据库协议解析那一步再考虑更重的方案也不会亏。2. 核心能力速览与API走读2.1 环境准备与安装安装只需要一条命令pip install sqlparse装完之后可以顺手确认下版本import sqlparse print(sqlparse.__version__)sqlparse 官方给出的定位是“无依赖、纯 Python”也就是你的虚拟环境里不会多出一堆连锁依赖。这点对开发调试非常友好尤其当你需要在生产环境快速部署一个临时脚本时体量小能省很多事。2.2 split把大脚本拆成单条语句sqlparse.split的逻辑很简单就是把一大段文本按语句边界切开返回一个字符串列表。它的智能之处在于能正确处理分号出现在字符串常量、注释、存储过程等场景中时不会被误切。import sqlparse sql select id, name from users where status 1; update users set status 0 where id 1; /* this is a comment; do not split here */ select count(*) from logs; statements sqlparse.split(sql) for i, s in enumerate(statements, 1): print(f语句{i}: {s[:50]}...)输出效果大致是语句1: select id, name from users where status 1; 语句2: update users set status 0 where id 1; 语句3: /* this is a comment; do not split here */ 语句4: select count(*) from logs;我实际用下来的感受是它对于大多数标准 SQL、存储过程脚本的分割都比较可靠这是手工正则很难做到的。2.3 parse返回Statement对象列表parse返回的是Statement对象列表每个Statement保留了结构化的 Token 分组信息。和split的区别在于parse不只是拿到纯文本片段而是能继续往下做结构化访问。parsed sqlparse.parse(sql) for stmt in parsed: print(stmt.get_type())get_type()会返回SELECT、UPDATE、CREATE这类首关键字非常适合拿来做语句类型分流。比如把输入的 SQL 分成 DDL、DML 两类再决定是否允许执行这在工单系统里很常用。还可以直接遍历Statement的tokens属性for stmt in parsed: for token in stmt.tokens: print(token.ttype, repr(token.value))这里会发现一个 Statement 是一个“容器”它内部有DML如SELECT、Keyword、Name、Whitespace、Punctuation等各种各样的 Token。掌握这个视角之后很多任务就变得像“在抽屉里按标签取东西”一样简单。2.4 format格式化SQL文本format是我日常用得最多的功能。它可以把一团乱 SQL 整理成风格统一、可读性高的样子。常用的参数有参数作用我的推荐值keyword_case关键字大小写upper或loweridentifier_case标识符大小写lower很多团队用strip_comments是否删除注释按需开启reindent是否重新缩进Truereindent_aligned对齐方式是否按列对齐False否则 diff 太大use_space_around_operators运算符两侧加空格Truecomma_first逗号是否前置看团队风格示例raw_sql select id,name,price from products where price100 order by id desc formatted sqlparse.format( raw_sql, keyword_caseupper, identifier_caselower, reindentTrue, use_space_around_operatorsTrue, ) print(formatted)输出SELECT id, name, price FROM products WHERE price 100 ORDER BY id DESC这种格式化在慢 SQL 分析、输出到日志、生成报表场景里非常顺手。你不要小看这几行代码它能省下大量肉眼比对 SQL 的时间。2.5 tokenize接口与Token结构如果你想从底层理解一条 SQL应该直接接触Token结构。sqlparse.tokenize会把文本切成一串 Token不按语句做进一步的归组适合做深度文本分析。from sqlparse import tokenize from sqlparse.tokens import Keyword, DML, String, Comment sql select * from users -- a comment for token in tokenize(sql): print(token.ttype, repr(token.value))常见 Token 类别包括Keywordselect、from、where、join等。DMLINSERT、UPDATE、DELETE这类数据操作关键字。Name表名、列名等标识符。String单引号或双引号包裹的字符串。Number数字。Comment注释包括--和/* */。还记得我们最初提到的正则痛点吗一旦有了 Token 类型按类型过滤就变得特别省心。比如想“去掉 SQL 里所有注释”from sqlparse import parse stmt parse(select * from users -- comment test)[0] clean_value .join( token.value for token in stmt.flatten() if token.ttype not in (Comment,) ) print(clean_value)flatten()会递归展开嵌套的分组 Token这个技巧在写各类 SQL 清洗工具时特别实用。3. 格式化与规范化实战3.1 快速实现“常量化”的SQL指纹工具做慢 SQL 优化时最头疼的事情之一是“同一条 SQL 只是条件值不同却被当成几百条记录”。这时候如果能把它们归并成一个指纹就方便统计 top N 相似 SQL。我的思路是先把 SQL 标准化用strip_comments去注释把字符串和数字替换成统一的占位符再调用format统一关键字大小写。这样同一类 SQL 的不同参数值就不会影响后续分组统计了。import sqlparse from sqlparse.tokens import String, Number def sql_fingerprint(sql_text): parsed sqlparse.parse(sqlparse.format( sql_text, strip_commentsTrue, keyword_caselower ))[0] tokens [] for token in parsed.flatten(): if token.ttype in (String, Number): tokens.append(?) else: tokens.append(token.value) normalized .join(tokens) normalized sqlparse.format( normalized, keyword_caselower, reindentTrue ) return normalized这个方案的核心是把“语义等价但字面量不同”的 SQL 映射到同一个文本串。实际跑在慢查询日志上原本看起来五花八门的 SQL可能最后只收敛到十几条核心模板优化方向立刻清晰很多。3.2 给慢SQL日志做标准化处理慢 SQL 日志往往是几百兆甚至上 GB 的大文件里面除了 SQL 本身还有时间、耗时、锁等待时间等元数据。用脚本逐行读出来再交给 sqlparse 去拆分和格式化是我处理这一类问题的标准姿势。import re import sqlparse slow_log_pattern re.compile( r# Query_time:\s([\d.])\s.*? r#\s r(.*?)(?# Query_time:|\Z), re.DOTALL ) def parse_slow_log(filepath): with open(filepath, r, encodingutf-8, errorsignore) as f: content f.read() for match in slow_log_pattern.finditer(content): query_time float(match.group(1)) sql_block match.group(2).strip() for statement in sqlparse.split(sql_block): formatted sqlparse.format( statement, keyword_caseupper, strip_commentsTrue, reindentTrue ) yield query_time, formatted这段逻辑其实是想说明一个通用步骤先用元信息把有效 SQL 段抽出来再用 sqlparse 拆成单条、格式化统一风格。格式化后的 SQL 方便写入中间表或者传给分析工具做聚合后续定位慢查询根因会轻松太多。3.3 借助pre-commit统一团队SQL风格我去年给团队做过一个很实用的东西在 git 的 pre-commit 阶段跑一个 py 脚本检查提交的.sql文件里的关键字大小写和基本缩进格式不符合规则就直接报错。实现思路就是用format后的文本和原文本比对。import sys import sqlparse def lint_sql_file(filepath): with open(filepath, r, encodingutf-8) as f: raw f.read() formatted sqlparse.format( raw, keyword_caseupper, identifier_caselower, strip_commentsFalse, reindentTrue, use_space_around_operatorsTrue, ) if formatted ! raw: print(fSQL风格不合格: {filepath}) return False return True if __name__ __main__: fail False for path in sys.argv[1:]: if path.endswith(.sql): fail not lint_sql_file(path) or fail sys.exit(1 if fail else 0)这里有个坑不要直接在钩子里自动改写文件否则会频繁制造大 diff队友容易心态爆炸。改成报告错误、让提交者手动修改效果反而更好。4. 解析能力用于安全检查与信息提取4.1 基于Statement类型分流在内部工单系统里经常需要区分用户提交的 SQL 是查询类还是变更类。用get_type()就能很快实现import sqlparse def classify_sql(sql_text): results [] for stmt in sqlparse.parse(sql_text): sql_type stmt.get_type().upper() if sql_type in (SELECT, SHOW, EXPLAIN): results.append((read, stmt)) elif sql_type in (INSERT, UPDATE, DELETE, MERGE, REPLACE): results.append((write, stmt)) elif sql_type in (CREATE, ALTER, DROP, TRUNCATE): results.append((ddl, stmt)) else: results.append((other, stmt)) return results有了这个分类可以在审批、审计、风险提示上做很多文章。比如 DDL 自动触发负责人审批SELECT类直接放行DELETE类强制二次确认。这不是危险功能而是把流程自动化减少人为疏忽。4.2 从SQL中提取表名提取表名看起来容易实际上一堆 SQL 风格差异很大有人写FROM users有人写FROM db.users u还有人写FROM (select ...) t。完整的表名抽取需要处理各种情况但 sqlparse 给了很好的起点。我用过的一个相对稳妥的简化策略import sqlparse from sqlparse.sql import Identifier, IdentifierList from sqlparse.tokens import Keyword, DML def extract_tables(sql_text): tables [] parsed sqlparse.parse(sql_text) for stmt in parsed: for token in stmt.tokens: if token.ttype in (DML, Keyword) and token.value.upper() in (FROM, JOIN, UPDATE, INTO): next_token token # 这里要拿到 FROM 后面的部分需要借助 sqlparse 对语句结构的识别 if isinstance(token, Identifier): tables.append(token.get_real_name()) elif isinstance(token, IdentifierList): for ident in token.get_identifiers(): tables.append(ident.get_real_name()) return tables实际上FROM后面的内容可能是Identifier、IdentifierList或一个括号子查询需要边界判断。更成熟的实现可以遍历 token 列表记录关键字的索引再取后续的标识符。这个函数不像官方库那样保证 100% 正确但结合正则和人工审查在文档生成、血缘分析场景已经能发挥大作用。4.3 辅助SQL注入风险检测聊到注入我只聊防御方向的思路。把用户输入的 SQL 绑定成参数化查询永远是第一位但很多遗留系统里确实还有拼接 SQL 的情况。写一个轻量检测工具把可能有风险的 SQL 挑出来告警是可行的。import sqlparse from sqlparse.tokens import Comment def has_risk(sql_text): parsed sqlparse.parse(sql_text) if not parsed: return False stmt parsed[0] # 检查注释很多绕过手法会在注释上做文章 for token in stmt.flatten(): if token.ttype is Comment: return True # 检测关键字组合 keywords [token.value.upper() for token in stmt.flatten() if token.ttype in sqlparse.tokens.Keyword] if UNION in keywords and SELECT in keywords: return True # 检测恒真条件如 OR 11 sql_upper stmt.value.upper() if OR 11 in sql_upper or OR 1 1 in sql_upper: return True return False这段代码只能覆盖很浅的特征千万不要把它当成万无一失的安全防线。真正安全还是要靠参数绑定、最小权限、SQL 白名单和源头规范化。sqlparse 在这里的角色是辅助过滤帮人早一步发现问题而不是替代防护体系。4.4 批量抽取DDL生成数据字典有一次我需要给几十张表自动整理成 Markdown 数据字典每张表结构都写一遍太耽误时间。我写了一个小脚本用 sqlparse 解析CREATE TABLE语句提取表名和字段清单再拼成表格输出。import sqlparse from sqlparse.tokens import Name, Punctuation def extract_columns(create_stmt): columns [] # 只解析 CREATE TABLE 语句的括号内部 for token in create_stmt.tokens: if isinstance(token, sqlparse.sql.Parenthesis): for subtoken in token.tokens: if subtoken.ttype is Name: columns.append(subtoken.value) return columns for stmt in sqlparse.parse(open(schema.sql).read()): if stmt.get_type() CREATE: table_name None # 通过 token 分析拿到表名 for token in stmt.tokens: if token.ttype is Name: table_name token.value break print(f表: {table_name}) print(extract_columns(stmt))这类脚本的价值不在技术难度而在于节省重复劳动。遇到成百上千张表时手写整理文档绝对是一场灾难。5. 踩坑清单与性能实测5.1 我踩过的4个坑第一split和parse的边界行为不完全一样。如果一段文本里有分号前后没有内容split()可能会过滤掉空串但parse()可能得到一个空的 Statement。写统计脚本时要注意空对象依然存在直接调get_type()可能返回空值或异常。第二format(reindentTrue)和format(reindent_alignedTrue)混用时输出可能和你预期的不一致。reindent_aligned想把对齐做得很细但自动生成的格式不一定符合团队习惯而且提交到 git 上会产生比较大的改动。建议先选定一种模式不要两个参数同时开。第三对特别复杂的大存储过程做格式化耗时会明显上升。sqlparse 虽然轻量但不是免费的午餐。它会将整条 SQL 放到内存里构建结构对于几十 MB 的超级脚本建议先split成单条再逐条处理避免一次解析过长文本导致内存抖动。第四get_type()不是万能的。遇到语法不完整的片段比如中途截断的 SQLget_type()可能返回UNKNOWN。不要盲目信任处理用户输入时要做好防御。5.2 性能和内存表现我用一台普通开发机8 核 16G处理过大约 10 万条常见 SQL 片段。每条平均几个关键字纯parse耗时总体在 10 秒量级如果只做format会更快一点。单条 SQL 如果特别长比如上千行的存储过程耗时可能在几十到几百毫秒。这类场景非常适合用多进程并行Python 的concurrent.futures.ProcessPoolExecutor就能直接上import concurrent.futures def process_one(sql_text): return sqlparse.format(sql_text, keyword_caseupper, reindentTrue) sql_list [...] # 从日志中切分好的 SQL with concurrent.futures.ProcessPoolExecutor(max_workers8) as executor: results list(executor.map(process_one, sql_list))实测中多进程能明显提升吞吐。注意不要在子进程里反复 import 过大量初始化的全局资源保持任务尽量纯函数化。5.3 命令行快速格式化工具最后分享一个能长期提升体验的小技巧把 sqlparse 封装成命令行工具随时粘贴 SQL 或者从文件输入格式化后的结果直接输出。cat messy.sql | python -c import sys, sqlparse sql sys.stdin.read() print(sqlparse.format(sql, keyword_caseupper, reindentTrue, use_space_around_operatorsTrue)) 你也可以把它写成一个.py脚本放到自己的工具目录里以后分析、分享 SQL 都方便很多。我在内部还把它做成了一个小小的缓存命令队友给我贴 SQL我格式化完再看效率高不少。sqlparse 这个库最大的价值是让你不必在正则和字符串拼接的泥潭里挣扎就能把 SQL 当结构化对象去处理。从批量格式化、指纹归并到分类提取、辅助风险识别它都是背后那个踏实好用的“隐形工具人”。如果你正好也想处理一堆 SQL 文本建议先拿split和format练手很快你会发现原来被字符串折磨的日子也可以这么清爽。
