# SQL 解析 API 服务（REST / HTTP）

基于 [sqlglot](https://github.com/tobymao/sqlglot) 的 SQL 解析 / 转译 / 格式化 / 提取 /
条件注入服务，使用 FastAPI 暴露为 REST 接口。核心逻辑在 `core.py`，被本服务（HTTP）复用。

## 运行

```bash
pip install -r requirements.txt
uvicorn main:app --host 0.0.0.0 --port 8000 --reload
```

启动后访问交互式文档：<http://localhost:8000/docs>

服务默认开启 CORS（`allow_origins=["*"]`），便于前端跨域调用。

## 接口总览

| 方法 | 路径 | 说明 |
| --- | --- | --- |
| GET | `/health` | 健康检查，返回 sqlglot 版本 |
| GET | `/dialects` | 列出 sqlglot 支持的所有方言 |
| POST | `/parse` | 将 SQL 解析为结构化 AST（JSON） |
| POST | `/transpile` | 在方言之间互转 SQL（如 MySQL → PostgreSQL） |
| POST | `/format` | 美化 / 标准化 SQL |
| POST | `/validate` | 校验 SQL 语法是否合法（不抛异常） |
| POST | `/extract` | 提取表名、列名、函数、CTE 及别名映射 |
| POST | `/add_condition` | 追加 WHERE 条件（自动别名，支持多表 / 多条件 / 多操作符 / 直拼原始 SQL） |

所有接口在 SQL 解析 / 处理失败时会返回 `400`，错误信息在响应体 `detail` 字段中。

---

## 接口详情

### 1. 健康检查 `GET /health`

返回服务状态与 sqlglot 版本号。

```bash
curl http://localhost:8000/health
# {"status":"ok","sqlglot_version":"30.12.0"}
```

### 2. 方言列表 `GET /dialects`

返回 sqlglot 支持的全部方言名称（按字母序）。

```bash
curl http://localhost:8000/dialects
# {"dialects":["bigquery","clickhouse","mysql","postgres", ...]}
```

### 3. 解析为 AST `POST /parse`

请求体：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 待解析的 SQL |
| `read` | string? | 输入方言，留空自动识别 |
| `pretty` | bool | AST 是否美化输出（默认 true） |

```bash
curl -X POST http://localhost:8000/parse \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a, b FROM t WHERE a > 1"}'
# {"sql":"...","dialect":null,"ast":"SELECT a, b FROM t WHERE a > 1","json":{...}}
```

返回含 `sql` / `dialect` / `ast` / `json` 四个字段，其中 `json` 为 AST 的树形结构。

### 4. 方言转译 `POST /transpile`

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 待转译的 SQL |
| `read` | string? | 输入方言，留空自动识别 |
| `write` | string | 目标方言（必填） |

```bash
curl -X POST http://localhost:8000/transpile \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT IFNULL(a, 0) FROM t", "write": "postgres"}'
# {"read":null,"write":"postgres","sql":"SELECT IFNULL(a, 0) FROM t","transpiled":"SELECT COALESCE(a, 0) FROM t"}
```

### 5. 格式化 `POST /format`

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 待格式化的 SQL |
| `read` | string? | 输入方言，留空自动识别 |
| `write` | string? | 输出方言，默认与输入一致（可顺带转译） |

```bash
curl -X POST http://localhost:8000/format \
  -H "Content-Type: application/json" \
  -d '{"sql": "select a,b from t where a>1"}'
# {"sql":"select a,b from t where a>1","formatted":"SELECT a, b\nFROM t\nWHERE a > 1"}
```

### 6. 语法校验 `POST /validate`

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 待校验的 SQL |
| `read` | string? | 输入方言，留空自动识别 |

返回不抛异常，`valid` 为布尔，`error` 在合法时为 `null`。

```bash
curl -X POST http://localhost:8000/validate \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM t"}'
# {"valid":true,"error":null}
```

### 7. 提取元数据 `POST /extract`

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 待分析的 SQL |
| `read` | string? | 输入方言，留空自动识别 |

返回字段：

| 字段 | 说明 |
| --- | --- |
| `tables` | 真实表名（已剔除 CTE 名） |
| `columns` | 带表限定符的列名，如 `t1.id` |
| `columns_bare` | 裸列名（不含来源），如 `id` |
| `functions` | 函数调用文本 |
| `ctes` | CTE 别名 |
| `table_aliases` | FROM/JOIN 上的显式别名映射 `[{alias, table}]` |
| `cte_sources` | CTE 名 → 内部真实表映射 `[{cte, tables}]` |

```bash
curl -X POST http://localhost:8000/extract \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a, b FROM t1 JOIN t2 ON t1.id = t2.id"}'
# {"tables":["t1","t2"],"columns":["a","b","t1.id","t2.id"],
#  "columns_bare":["a","b","id"],"functions":[],"ctes":[],
#  "table_aliases":[],"cte_sources":[]}
```

### 8. 追加条件 `POST /add_condition`

为 SQL 追加 WHERE 条件，支持 **多表 / 多条件 / 多操作符 / 直拼原始 SQL**。自动定位每个真实表在 SQL 中的引用名（别名优先），把裸列或真实表名限定列改写为别名前缀后，注入到「该表所属查询的作用域」。

提供三种入参方式（互斥，优先级：`conditions` > `raw` > 兼容旧版）：

**方式一 · 结构化（推荐，支持多表 + 多操作符）** — 字段 `conditions` 为条件数组，每个元素：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `column` | string | 列名，可带引用前缀，如 `"status"` 或 `"t1.id"` |
| `op` | string | 操作符，默认 `=`。支持 `= != > >= < <= LIKE IN NOT IN BETWEEN IS NULL IS NOT NULL` |
| `value` | any? | 比较值；`IN` / `NOT IN` 用列表，`BETWEEN` 用 `[下限, 上限]` 两元素列表，`IS NULL` 可省略 |
| `table` | string? | 可选，条件所属真实表名（多表场景）；跨表 JOIN 条件留空，注入顶层查询 |
| `combine` | string | 与上一条条件的连接符 `AND`（默认）或 `OR` |

```bash
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{
    "sql": "SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id WHERE t1.flag = 0",
    "conditions": [
      {"column": "status", "op": "=",   "value": 1, "table": "table1", "combine": "AND"},
      {"column": "amount", "op": ">",   "value": 100, "table": "table1"},
      {"column": "id",     "op": "IN",  "value": [1, 2, 3], "table": "table2", "combine": "OR"},
      {"column": "t1.id",  "op": "=",   "value": "t2.fid"}
    ]
  }'
# SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id
# WHERE t1.flag = 0 AND (((t1.status = 1 AND t1.amount > 100) OR t2.id IN (1, 2, 3)) AND t1.id = 't2.fid')
```

**方式二 · 直拼原始 SQL（自动把真实表名替换为别名）** — 字段 `raw`，直接写 WHERE 片段（用真实表名），系统自动改写为别名后注入：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `raw` | string \| string[] | 原始 WHERE 片段（用真实表名写）；多条按 `operator` 合并 |
| `operator` | string | 多 raw 片段之间的合并方式，`AND`（默认）或 `OR` |
| `read` | string? | 输入方言，留空自动识别 |

```bash
# 片段里用真实表名 table1 / table2 写，自动替换为别名 t1 / t2
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id WHERE t1.flag = 0",
       "raw": "table1.status = 1 AND table2.amount > 100"}'
# SELECT a FROM table1 AS t1 JOIN table2 AS t2 ON t1.id = t2.id
# WHERE t1.flag = 0 AND (t1.status = 1 AND t2.amount > 100)

# 多条 raw 片段按 operator 合并
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1",
       "raw": ["table1.status = 1", "table1.deleted_at IS NULL"], "operator": "OR"}'
# SELECT a FROM table1 AS t1 WHERE table1.status = 1 OR table1.deleted_at IS NULL
```

**方式三 · 兼容旧版（单表 + 原始条件串）**：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `sql` | string | 原始 SQL |
| `table` | string | 真实表名（非别名），用于定位引用名 |
| `condition` | string \| string[] | 条件片段；单条字符串或字符串数组（多条件） |
| `operator` | string | 多条件合并方式，`AND`（默认）或 `OR` |
| `read` | string? | 输入方言，留空自动识别 |

```bash
curl -X POST http://localhost:8000/add_condition \
  -H "Content-Type: application/json" \
  -d '{"sql": "SELECT a FROM table1 AS t1", "table": "table1",
       "condition": ["status = 1", "amount > 100"], "operator": "AND"}'
# SELECT a FROM table1 AS t1 WHERE t1.status = 1 AND t1.amount > 100
```

**选型建议：**

- 需要精确控制**多表分作用域**和**每个条件的操作符** → 用 `conditions`（方式一）。
- 已有现成的 WHERE 片段（用真实表名写）想直接拼上去 → 用 `raw`（方式二）。
- 老接口 / 单表简单场景 → 用 `table` + `condition`（方式三）。

**行为说明：**

- 自动建立「真实表名 → 别名」映射；`raw` 直拼片段里用真实表名写的列（如 `table1.status`）会自动改写为别名（`t1.status`）。
- 裸列自动加引用名前缀（`status = 1` → `t1.status = 1`）；列已带前缀（如 `t1.id`）则原样保留。
- **多表**：条件按 `table` 定位到各自所属查询作用域；表位于子查询 / CTE 内部时，条件只注入到那一层。所有表在同一查询（如 JOIN）时，条件合并进同一 WHERE。`raw` 片段若跨多表，注入到这些表的最近公共查询作用域。
- **多操作符**：除 `AND` / `OR` 连接符外，每个条件自带 `op`（`IN` / `BETWEEN` / `LIKE` / `IS NULL` 等）。
- 同一作用域内多条件按各自 `combine` 从左到右联结；若目标查询已有 WHERE，则与原有条件 `AND` 合并。
- 同一真实表在 SQL 中出现多次时取第一个匹配；找不到表 / 操作符不合法 / SQL 解析失败时返回 `400` 错误。

---

## 错误处理

所有接口在 SQL 解析 / 处理失败时会返回 HTTP `400`，错误信息位于响应体的 `detail` 字段：

```json
{ "detail": "解析失败: ..." }
```

常见错误与含义：

| `detail` 片段 | 含义 |
| --- | --- |
| `解析失败: ...` | 输入 SQL 或条件片段无法被 sqlglot 解析（语法错误） |
| `未在 SQL 中找到表: xxx` | `table` / `conditions[].table` 指定的真实表在 SQL 中不存在 |
| `条件不能为空` | 传入的 `condition` / `raw` 为空 |
| `不支持的合并方式: xxx` | `operator` 不是 `AND` / `OR` |
| `不支持的操作符: xxx` | `conditions[].op` 不在支持的操作符列表内 |
| `IN 操作符的 value 必须为列表` | 使用 `IN` / `NOT IN` 时 `value` 不是数组 |
| `BETWEEN 操作符的 value 必须为包含 [下限, 上限] 的列表` | 使用 `BETWEEN` 时 `value` 不是两元素数组 |

> 提示：`/validate` 接口不会抛异常，可通过其 `valid` / `error` 字段预先判断 SQL 是否合法，再调用其它写操作接口。

## MCP 服务（供 Agent 调用）

除了 REST 接口，本服务还提供 **MCP Server**（`mcp_server.py`），让 Agent 通过 MCP 协议（stdio 传输）直接调用 SQL 解析/校验/权限能力。

特点：**零额外依赖**，仅依赖 `sqlglot`（未使用官方 `mcp` SDK 分叉包，避免版本/兼容风险），通过标准「换行分隔的 JSON-RPC 2.0」与任何 MCP 客户端（Claude Desktop、自定义 Agent 等）对接。

运行（Agent 以子进程方式拉起）：

```bash
python mcp_server.py
```

### 暴露的工具

| 工具 | 说明 |
| --- | --- |
| `validate_sql` | 校验 SQL 语法是否合法，返回 `{valid, error}`（不抛异常） |
| `parse_sql` | 将 SQL 解析为结构化 AST（JSON） |
| `transpile_sql` | 方言互转（如 MySQL → PostgreSQL） |
| `format_sql` | 美化 / 标准化 SQL |
| `extract_sql` | 提取表名、列名、函数、CTE 及别名映射 |
| `add_condition` | 追加 WHERE 条件（多表 / 多条件 / 多操作符 / 直拼原始 SQL） |
| `check_permission` | 按策略校验 SQL（表/列白黑名单、禁止 DML/DDL 等），**仅校验不改写** |
| `enforce_permission` | 校验通过后自动加固 SQL（注入行级过滤、自动补 LIMIT），返回改写结果 |

### 权限策略（policy）

`check_permission` / `enforce_permission` 共用如下策略（全部可选）：

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| `allow_write` | bool | 是否允许 DML（INSERT/UPDATE/DELETE），默认 `false` |
| `allow_ddl` | bool | 是否允许 DDL（CREATE/DROP/ALTER/TRUNCATE），默认 `false` |
| `allowed_tables` | list | 表白名单（非空时，不在其中的表一律拒绝） |
| `denied_tables` | list | 表黑名单（命中即拒绝，优先级高于白名单） |
| `allowed_columns` | dict | 列级白名单 `{表: [列,...]}`（仅校验带表限定符的列） |
| `row_filters` | dict | 行级过滤 `{表: "条件"}`，`enforce_permission` 自动注入 WHERE |
| `max_limit` | int | SELECT 自动补 LIMIT 上限（无 LIMIT 时） |

`enforce_permission` 示例（`enforce_permission` 工具）：

```json
{
  "sql": "SELECT id, amount FROM orders",
  "policy": {
    "allow_write": false,
    "row_filters": { "orders": "tenant_id = 7" },
    "max_limit": 100
  }
}
```

→ 返回 `SELECT id, amount FROM orders WHERE tenant_id = 7 LIMIT 100`（行级隔离 + 限流已自动生效）。

### Agent 接入配置示例

**Claude Desktop / 兼容 MCP 的客户端**（`claude_desktop_config.json`）：

```json
{
  "mcpServers": {
    "sqlglot": {
      "command": "python",
      "args": ["/绝对路径/mcp-sqlglot/http/mcp_server.py"]
    }
  }
}
```

**自定义 Agent（stdio 调用骨架）**：向子进程 stdin 逐行写入 JSON-RPC 消息，读取 stdout 每行响应：

```text
→ {"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"agent","version":"1"}}}
← {"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2024-11-05","capabilities":{"tools":{}},"serverInfo":{"name":"sqlglot-mcp","version":"1.0.0"}}}
→ {"jsonrpc":"2.0","id":2,"method":"tools/list"}
→ {"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"enforce_permission","arguments":{"sql":"SELECT * FROM orders","policy":{"row_filters":{"orders":"tenant_id = 7"},"max_limit":100}}}}
```

---

```
http/
├── main.py          # FastAPI 应用与全部 REST 接口
├── mcp_server.py    # MCP Server（stdio JSON-RPC，供 Agent 调用，零额外依赖）
├── core.py          # 核心逻辑（解析/转译/格式化/提取/条件注入/权限校验）
├── requirements.txt # 依赖（fastapi, uvicorn, sqlglot, pydantic, ...）
└── README.md        # 本文档
```