安全执行 SQL 并返回查询结果

这是一篇系列文，请按顺序阅读。话说回来，

这篇文章目标

前面的节点已经完成：

使用者问题 → 召回字段、指标和维度值 → 过滤候选 → 组装 SQL 上下文 → 生成 SQL → 验证并有限修正 SQL

最终还差一步这方面。执行已经的 SQL，并把数据库结果转换成前端可以通过 SSE 接收的 JSON 数据。

使用者痛点：

• 返回几十万行，把内存和 SSE 响应撑大。不过，
• 扫描大量数据。长时间占用数据库连接，
• 返回 Decimal日期或二进制数据，导致 json.dumps 失败。
• 出现同名输出列，转换成字典时互相覆盖。
• 因数据库状态变化，在真正执行时失败。
• 因未来图结构调整，错误地绕过验证节点。

这篇文章实现的受控执行流程：

validate_sql
│
│ sql_validated = true
↓ run_sql
检查验证标记
↓ SQLGlot 添加 max_rows + 的内部 LIMIT
↓ SET TRANSACTION READ ONLY
↓ 在应用超时内执行查询
↓ 最多读取 max_rows + 行
↓ 回滚只读事务
↓ 序列化 Decimal / 日期 / bytes / 长文本
↓ 通过 SSE 返回结果、截断信息和耗时

先看输入和输出

. run_sql 的输入

执行节点从 LangGraph State 读取：

{ "sql": """ SELECT SUM AS sales_amount FROM fact_order AS fo JOIN dim_region AS dr ON fo.region_id = dr.region_id WHERE dr.province = '北京市' """。
"sql_validated": True,"validation_errors":,"error": None,}

其中真正决定是否可以进入执行节点的是：

sql_validated is True and error is None

不能只判断 error is None因为工作流初始 State 也可能暂时没有错误，但此时 SQL 还没有。

. 查询成功时的 State 输出

{ "execution_result": { "columns":,"rows":,"row_count":,"truncated": False,"truncated_cell_count":。"elapsed_ms":,},"error": None,}

字段含义：

• columns: 按数据库返回顺序排列的列名。
• rows: 转换成 JSON 兼容类型的结果行。
• row_count: 实际返回给前端的行数。
• truncated: 是否因程序行数上限丢弃了更多结果行。不过，
• truncated_cell_count: 有多少个超长单元格被截断。
• alert_ms: 从开始执行到完成序列化前的近似毫秒数。

金额使用字符串返回是有意设计：

{"sales_amount": ""}

If MySQL DECIMALs were forced into JavaScript floating‑point numbers。precision would be lost.

. 查询成功时的 SSE 输出

{
说到"type"，"result","columns":,"data":,"row_count":,"truncated": false,"truncated_cell_count":,"elapsed_ms":
}

The data field keeps existing front‑end consumption unchanged;extra metadata lives alongside it in same event.

. 查询失败时的输出

If MySQL disconnects,schema changes mid‑flight,or query times out:

{
从"type"来看，"execution_error","message": "MySQL 执行失败：..."
}

The State records:

{ "execution_result": {},"error": "MySQL 执行失败：...",}

This error does **not** trigger anor LLM round – runtime failures are usually caused by connection/permission/concurrency issues that cannot be fixed by rewriting SQL.

. 没有验证标记时的输出

If someone calls run_sql directly without flag:

{'sql_validated': True}

The node returns:

SQL 尚未，拒绝执行

This second defensive check lives outside graph routing.

第一部分这方面，项目实现

.设计执行配置

Add to conf/app_config.yaml:

sql_execution:
max_rows:
timeout_seconds:
max_cell_chars:

The three dimensions are:

• max_rows – maximum rows sent to front‑end
• timeout_seconds – maximum seconds app will wait for MySQL
• max_cell_chars – per‑cell character limit for text/BLOB values

Pydantic Settings counterpart:

class SQLExecutionSettings:
max_rows: int = Field
timeout_seconds: float = Field
max_cell_chars: int = Field

The defaults mselves have safe boundaries to avoid accidental “unlimited” settings.

. 为什么 Repository 和 Service 要分开

• Repository – raw MySQL connection handling .
• Service – application‑level concerns .

This separation enables unit tests that mock repository while exercising all Python‑side logic.

. 定义 Repository 原始结果

@dataclass
class RawSQLExecutionResult:
columns: list
再看rows。list]
truncated: bool

. 使用 AST 添加内部 LIMIT

return expression.limit.sql

. 不扩大使用者原本的 LIMIT

• No LIMIT → system adds LIMIT {fetch_limit} → client sees at most {max_rows}=1000 rows. truncated flag may be true.
• Original LIMIT ≤ fetch_limit → keep original LIMIT unchanged.
CTE & UNION also get a top‑level LIMIT instead of being added to inner subqueries.

. 执行前 解析 SQL

. 开启 MySQL 只读事务

async with self.database.engine.connect as connection:
await connection.exec_driver_sql
result = await connection.exec_driver_sql

• AST validation forbids non‑SELECT statements.
• run_sql requires `sql_validated=True`.
• MySQL session set to READ ONLY.
• Database account granted only SELECT privileges.

. 查询成功也回滚  ​​​​​​​​​​

if connection.in_transaction:
await connection.rollback

. 使用 execdriversql  ​​​​​

. 设置应用等待超时 ​​​

async with asyncio.timeout:
raw_result = await self.repository.execute

...

。

---