高性能SQL解析库-fast-sqlparse

原本是我写的一个C++ 17跨平台SQL解析库，后面用pybind11编译成了pyd和so文件，然后二次开发而来，他的速度有一定的损失，但是我们解析SQL更简单、更快、更直观了。经过一年7个大版本的迭代开发、反复测试和不断完善，今年我把它发布到github上，希望有人能看到、使用、提出问题和意见。
相比与纯python库sqlparse、sqlglot等它的速度通常会快出数十倍。另外它还非常适合解析超大、复杂查询、子查询深度嵌套、公共表达式混用的场景。

GitHub仓库

https://github.com/Nohaltsail/fast-pysqlparse

目录

• 安装
• 快速开始
• 核心类说明
• 功能演示
• 性能对比
• API参考

安装

pipinstallfast-pysqlparse

快速开始

fromfastsqlparseimportParsed,ParsedQuery# 解析SQLsql="SELECT * FROM users WHERE age > 18"parsed=Parsed(sql)# 获取解析结果query=parsed.parsedforest[0]print(query.sources)# 数据源print(query.columns)# 列信息print(query.format())# 格式化输出

核心类说明

1. Parsed - SQL解析器主类

功能: 解析任意SQL语句（SELECT、INSERT、UPDATE、DELETE、CREATE等）

参数:

• sql_statements(str): SQL语句字符串
• file(str, optional): SQL文件路径
• name(str, optional): 解析内容名称
• pure(bool, default=False): 是否忽略注释

主要属性和方法:

• parsedforest: 返回解析后的语句列表
• statements: 所有SQL语句
• tokens(): 获取词法单元
• AST(): 获取抽象语法树（JSON格式）
• format(indent): 格式化SQL
• content(): 获取原始SQL内容
• name: SQL语句名称

示例:

fromfastsqlparseimportParsed sql="SELECT u.id, u.name FROM users u WHERE u.age > 18"parsed=Parsed(sql)# 获取解析树items=parsed.parsedforest# 格式化formatted=parsed.format(indent=" ")# 获取ASTast_json=parsed.AST()# 获取Tokenstokens=parsed.tokens()

2. ParsedQuery - SELECT查询解析器

功能: 专门解析SELECT查询语句，提取查询子句和元数据

参数:

• statement(str): SELECT语句
• name(str): 查询名称
• pure(bool, default=False): 是否去除注释

主要属性:

• sources: 数据源列表（FROM/JOIN的表）
• columns: 选择的列列表
• clause_select: SELECT子句内容
• clauses: 子句列表
  • FROM子句内容
  • WHERE子句内容
  • GROUP BY/HAVING子句
  • ORDER BY子句
  • LIMIT子句
• parent: Parsed父对象
• cte: CTE映射字典
• unions: UNION查询列表
• subquery: 子查询信息
• level: 嵌套层级

主要方法:

• format(indent, init_indent): 格式化查询
• ast(): 生成AST
• tokens(): 获取Tokens
• tokenize(statement): 静态方法，快速词法分析

示例:

fromfastsqlparseimportParsedQuery sql=""" SELECT u.user_id, COUNT(o.order_id) as cnt FROM users u LEFT JOIN orders o ON u.user_id = o.user_id WHERE u.status = 'active' GROUP BY u.user_id HAVING cnt > 5 ORDER BY cnt DESC LIMIT 10 """query=ParsedQuery(sql,"user_orders")# 提取信息print("数据源:",query.sources)print("列:",query.columns)fori,clauseinenumerate(query.clauses):ifclause.part=="CLAUSE_FROM":print(f"FROM子句:{clause.clause}")elifclause.part=="CLAUSE_WHERE":print(f"WHERE条件:{clause.clause}")elifclause.part=="CLAUSE_AGGREGATION":print(f"GROUP BY:{clause.clause}")elifclause.part=="CLAUSE_SORT":print(f"ORDER BY:{clause.clause}")elifclause.part=="CLAUSE_LIMIT":print(f"LIMIT:{clause.clause}")# 快速tokenizertokens=ParsedQuery.tokenize(sql)fortoken_type,token_value,posintokens[:5]:print(f"{token_type}:{token_value}")

3. ParsedCTE - 公用表表达式解析器

功能: 解析WITH子句（CTE）

参数:

• statement(str): WITH语句
• pure(bool, default=False): 是否去除注释
• name(str, optional): CTE名称

主要属性:

• raw: 原始CTE语句
• cte_stmts: CTE语句列表
• name: CTE名称

主要方法:

• format(indent, init_indent): 格式化CTE
• ast(): 生成AST
• tokenize(statement): 静态方法，快速词法分析

示例:

fromfastsqlparseimportParsedCTE,ParsedQuery sql=""" WITH RECURSIVE cte AS ( SELECT 1 as n UNION ALL SELECT n + 1 FROM cte WHERE n < 10 ) """cte=ParsedCTE(sql)print("CTE语句:",cte.cte_stmts)print("格式化:\n",cte.format())sql=""" WITH RECURSIVE cte AS ( SELECT 1 as n UNION ALL SELECT n + 1 FROM cte WHERE n < 10 ) SELECT * FROM cte """ctes=ParsedQuery(sql,'test').cteforcte_nameinctes:print("CTE名称:",cte_name)print("CTE语句:",ctes[cte_name].format())

4. ParsedInsert - INSERT语句解析器

功能: 解析INSERT语句，支持VALUES和SELECT两种方式

参数:

• statement(str): INSERT语句
• pure(bool, default=False): 是否去除注释

主要属性:

• name: 目标表名
• columns: 插入的列列表
• values: 插入的值
• query: 查询对象（INSERT…SELECT时）
• query_load: 是否有查询加载
• main_stmt: 主语句
• cte_stmt: CTE语句
• query_stmt: 查询语句

主要方法:

• format(indent, init_indent): 格式化
• ast(): 生成AST
• tokens(): 获取Tokens
• tokenize(statement): 静态方法，快速词法分析

示例:

fromfastsqlparseimportParsedInsert sql1="INSERT INTO users (id, name) VALUE (1, 'Alice')"insert1=ParsedInsert(sql1)print("表名:",insert1.name)print("列:",insert1.columns)print("值:",insert1.values)# SELECT方式（带CTE）sql2=""" INSERT INTO summary (product_id, total) WITH stats AS ( SELECT product_id, SUM(amount) as total FROM orders GROUP BY product_id ) SELECT product_id, total FROM stats sts """insert2=ParsedInsert(sql2)print("表名:",insert2.name)print("有查询:",insert2.query_load)ifinsert2.query:forsourceininsert2.query.sources:print("子句:",source.raw)print("表:",source.table)print("别名:",source.alias)

5. 其他解析器类

ParsedView - VIEW解析器

fromfastsqlparseimportParsedView sql="CREATE VIEW active_users AS SELECT * FROM users WHERE status='active'"view=ParsedView(sql)

ParsedUpdate - UPDATE解析器

fromfastsqlparseimportParsedUpdate sql="UPDATE users SET status='inactive' WHERE last_login < '2023-01-01'"update=ParsedUpdate(sql)

ParsedDelete - DELETE解析器

fromfastsqlparseimportParsedDelete sql="DELETE FROM logs WHERE created_at < '2023-01-01'"delete=ParsedDelete(sql)

ParsedCreate - CREATE TABLE解析器

fromfastsqlparseimportParsedCreate sql=""" CREATE TABLE users ( id INT PRIMARY KEY, name VARCHAR(100), email VARCHAR(200) ) """create=ParsedCreate(sql)

功能演示

场景1: 普通查询（含子查询）

fromfastsqlparseimportParsed sql=""" SELECT u.user_id, u.username, (SELECT COUNT(*) FROM orders o WHERE o.user_id = u.user_id) as order_count FROM users u WHERE u.age > 18 ORDER BY u.username LIMIT 10 """parsed=Parsed(sql)query=parsed.parsedforest[0]# 提取关键信息print("数据源:",query.sources)print("列:",query.columns)print("SELECT子句:",query.clause_select)forclauseinquery.clauses:ifclause.part=="CLAUSE_FROM":print(f"FROM子句:{clause.clause}")elifclause.part=="CLAUSE_WHERE":print(f"WHERE子句:{clause.clause}")elifclause.part=="CLAUSE_SORT":print(f"ORDER BY子句:{clause.clause}")elifclause.part=="CLAUSE_LIMIT":print(f"LIMIT子句:{clause.clause}")

输出:

数据源: [<DqlSourceExpr object>] 列: [<DqlColumnExpr object>, ...] SELECT子句: ['u.user_id', 'u.username', '(SELECT COUNT(*) ...) as order_count'] FROM子句: FROM users u WHERE子句: WHERE u.age > 18 ORDER BY子句: ORDER BY u.username LIMIT子句: LIMIT 10

场景2: 临时结果集（聚合查询）

fromfastsqlparseimportParsedimportjson sql=""" WITH sales_summary AS ( SELECT product_id, SUM(amount) as total_sales, AVG(amount) as avg_sales FROM sales WHERE sale_date >= '2024-01-01' GROUP BY product_id ) SELECT * FROM sales_summary WHERE total_sales > 1000 """parsed=Parsed(sql)# 获取Tokenstokens=parsed.tokens()print(f"Token数量:{len(tokens)}")# 获取ASTast_str=parsed.AST()ast_obj=json.loads(ast_str)print(json.dumps(ast_obj,indent=2,ensure_ascii=False))

场景3: UNION查询 + Tokenizer

fromfastsqlparseimportParsedQuery sql=""" WITH region_sales AS ( SELECT region, SUM(amount) as total FROM sales GROUP BY region ) SELECT * FROM region_sales UNION ALL SELECT 'TOTAL' as region, SUM(total) FROM region_sales """# 使用Tokenizer进行快速词法分析tokens=ParsedQuery.tokenize(sql)fortoken_type,token_value,positionintokens:print(f"Type:{token_type:15}| Value:{token_value[:30]:30}| Pos:{position}")

场景4: INSERT INTO … CTE SELECT

fromfastsqlparseimportParsedInsert sql=""" INSERT INTO summary_table (product_id, total_amount, avg_amount) WITH product_stats AS ( SELECT product_id, SUM(amount) as total_amount, AVG(amount) as avg_amount FROM orders GROUP BY product_id ) SELECT product_id, total_amount, avg_amount FROM product_stats """insert=ParsedInsert(sql)print("目标表:",insert.name)print("插入列:",insert.columns)print("有查询:",insert.query_load)ifinsert.query:print("查询类型:",type(insert.query))print("查询来源:",insert.query.sources)print("查询列:",insert.query.columns)

场景5: 处理注释和格式化

fromfastsqlparseimportParsed,strip_note sql=""" -- 这是主注释 SELECT u.user_id, -- 用户ID u.username -- 用户名 FROM users u /* 用户表 */ WHERE u.status = 'active' -- 只查活跃用户 """# 保留注释并格式化parsed_with_comments=Parsed(sql,pure=False)print("保留注释:")print(parsed_with_comments.format())# 去除注释并格式化parsed_pure=Parsed(sql,pure=True)print("\n去除注释:")print(parsed_pure.format())# 仅去除注释（不格式化）stripped=strip_note(sql)print("\n仅去注释:")print(stripped)

性能对比

测试环境

• SQL长度: 1359字符
• 测试次数: 100次

性能结果

大规模测试

测试1: 5000次解析

• SQL长度: 639字符
• 总耗时: 0.6084秒
• PPS (Parses Per Second): 8218.88
• 平均每次: 0.1217ms

测试2: 1000万字符SQL

• SQL长度: 10,500,998字符
• 总耗时: 1.4085秒
• CPS (Characters Per Second): 7,455,540
• 解析成功！

API参考

工具函数

strip_note(sql: str) -> str

去除SQL中的注释

fromfastsqlparseimportstrip_note sql="SELECT * FROM users -- comment"clean=strip_note(sql)# 结果: "SELECT * FROM users"

format(sql: str, indent: str = " ") -> str

格式化SQL语句

fromfastsqlparseimportformatsql="SELECT * FROM users WHERE id=1"formatted=format(sql,"query",indent=" ")

tokenize(sql: str) -> List[Tuple[str, str, int]]

词法分析

tokenize_query(sql: str) -> List[Tuple[str, str, int]]

快速词法分析SELECT语句

fromfastsqlparseimporttokenize_query tokens=tokenize_query("SELECT * FROM users")

tokenize_cte(sql: str) -> List[Tuple[str, str, int]]

快速词法分析WITH语句

tokenize_insert(sql: str) -> List[Tuple[str, str, int]]

快速词法分析INSERT语句

tokenize_update(sql: str) -> List[Tuple[str, str, int]]

快速词法分析UPDATE语句

tokenize_delete(sql: str) -> List[Tuple[str, str, int]]

快速词法分析DELETE语句

tokenize_view(sql: str) -> List[Tuple[str, str, int]]

快速词法分析VIEW语句

Token结构

每个Token包含以下属性:

• type: Token类型（KEYWORD, IDENTIFIER, LITERAL, WHITESPACE等）
• value: Token的值
• position: 在SQL中的位置

tokens=parsed.tokens()fortokenintokens:print(f"Type:{token.type}, Value:{token.value}, Pos:{token.at}")

AST结构

AST以JSON格式返回，包含:

• 查询子句（SELECT, FROM, WHERE等）
• CTE定义
• 列信息
• 数据源信息
• 联合查询信息

importjson ast_json_list=parsed.AST()ast_json_dic=parsed_query.ast()ast_obj=json.loads(ast_json_dic)

最佳实践

1. 选择合适的解析器

• 通用SQL: 使用Parsed
• 仅SELECT: 使用ParsedQuery（更快）
• 仅INSERT: 使用ParsedInsert
• 仅CTE: 使用ParsedCTE

2. 性能优化

• 如果只需要词法信息，使用tokenize()静态方法
• 设置pure=True可以跳过注释处理，提升速度
• 避免重复解析相同SQL，缓存解析结果

3. 错误处理

fromfastsqlparseimportParsedtry:parsed=Parsed(invalid_sql)exceptExceptionase:print(f"解析失败:{e}")

常见问题

Q1: 如何提取表名？

query=parsed.parsedforest[0]forsourceinquery.sources:print(source.table)# 或查看source的属性

Q2: 如何处理多语句SQL？

parsed=Parsed("SELECT * FROM t1; SELECT * FROM t2;")forstmtinparsed.parsedforest:print(stmt)

Q3: 如何获取子查询信息？

query=parsed.parsedforest[0]ifquery.subquery:forsubqinquery.subquery:print(subq)