mongo-2-sql 0.1.4
A library to convert MongoDB aggregation pipelines to SQL parser
License: MIT License
Requires Python: >=3.10

# Mongo to SQL

Mongo to SQL 是一个将 MongoDB 聚合管道转换为 SQL 查询语句的工具。

## 版本更新 v0.1.4

### 新增功能

#### 1. CTE（公用表表达式）统一处理
- **$addFields、$project、$group 阶段现在使用 CTE 方式处理**
- 将复杂的聚合管道拆分为多个SQL片段，使用 WITH 子句拼接
- 自动管理 CTE 命名和依赖关系
- 适用于多次投影、多层分组等复杂场景

#### 2. 表达式解析器统一架构
- **ExpressionResolver**: 统一的表达式解析器入口，避免 stage 之间相互 import
- **BaseExpressionParser**: 基础表达式解析器基类，减少代码重复
- 提供通用操作符集合、字段引用解析、值转换等基础功能

#### 3. SQL 格式化输出
- 使用 sqlparse 库自动格式化 SQL
- SELECT、FROM、WHERE、GROUP BY 等关键字独占一行
- 字段列表缩进对齐，提高可读性
- WHERE 子句添加注释提示可扩展性

#### 4. Project 阶段增强
- 支持字段重命名语法: `{"new_name": "$old_name"}`
- 聚合后字段引用自动处理（无需添加表前缀）
- 支持包含模式和排除模式

#### 5. Stage 解耦架构
- 各 stage 表达式解析器通过统一的 ExpressionResolver 进行交互
- 避免模块之间的循环依赖
- 更易于扩展和维护

### 问题修复

- **修复 `_id` 字段双引号问题**: 不再使用 `"_id"`，改为 `_id`
- **修复聚合后字段引用**: 聚合后的字段不再错误添加表前缀
- **修复模块导入路径**: 统一使用相对导入，避免 "attempted relative import beyond top-level package" 错误

## 功能特性

- **阶段基础方法支持**: 支持 `$match`, `$project`, `$group`, `$sort`, `$limit`, `$lookup`, `$unwind` 等常用阶段
- **丰富的操作符**: 支持比较操作符、逻辑操作符、聚合函数、条件表达式、字符串函数等
- **可扩展架构**: 模块化的设计，易于添加新的阶段处理器和自定义函数
- **多方言支持**: 主要支持 SQLite，可扩展支持 PostgreSQL、MySQL 等 SQL 方言
- **类型安全**: 完整的类型注解支持，提供良好的 IDE 提示
- **格式化输出**: 自动生成格式化的 SQL，提高可读性
- **错误处理**: 完善的错误检查和友好的错误提示
- **性能优化**: 针对不同 SQL 方言的特定优化

## 安装

```bash
pip install mongo-2-sql
```

## API 文档

### 主要函数

#### `convert_mongo_pipeline_to_sql(pipeline, collection_name, dialect='sqlite', validate=True)`

将 MongoDB 聚合管道转换为 SQL 查询。

**参数:**
- `pipeline` (List[Dict[str, Any]]): MongoDB 聚合管道
- `collection_name` (str): MongoDB 集合名称（对应 SQL 表名）
- `dialect` (str): SQL 方言，默认为 'sqlite'
- `validate` (bool): 是否验证管道格式，默认为 True

**返回:**
- `(str, Dict[str, Any])`: (SQL 查询字符串, 元数据字典)

**异常:**
- `ValueError`: 管道格式无效
- `RuntimeError`: 转换过程中发生错误

#### `MongoToSQLConverter(dialect='sqlite', stage_loader=None)`

面向对象的转换器类。

**方法:**
- `convert(pipeline, collection_name, validate=True)`: 转换聚合管道
- `convert_single_stage(stage, collection_name)`: 转换单个阶段
- `get_supported_stages()`: 获取支持的阶段列表
- `is_stage_supported(stage_name)`: 检查阶段是否受支持

### 便捷函数

- `mongo_match_to_sql(match_spec, table_name='t')`: 转换 $match 阶段
- `mongo_project_to_sql(project_spec, table_name='t')`: 转换 $project 阶段
- `mongo_group_to_sql(group_spec, table_name='t')`: 转换 $group 阶段

## 快速开始

### 基本用法

```python
from mongo_2_sql import convert_mongo_pipeline_to_sql

# 定义 MongoDB 聚合管道
pipeline = [
    { "$match": { "status": "active", "age": { "$gte": 18 } } },
    { "$group": { "_id": "$category", "count": { "$sum": 1 } } },
    { "$sort": { "count": -1 } },
    { "$limit": 10 }
]

# 转换为 SQL
sql, metadata = convert_mongo_pipeline_to_sql(pipeline, "users")
print(sql)
```

输出：
```sql
SELECT 
    t.category AS _id,
    COUNT(*) AS count
FROM users AS t
WHERE 1=1
 AND (t.status = 'active')
 AND (t.age >= 18)
GROUP BY t.category
ORDER BY count DESC
LIMIT 10
```

### CTE 处理（复杂聚合管道）

当管道包含多次 $project、$group 或 $addFields 阶段时，系统会自动使用 CTE 方式处理：

```python
from mongo_2_sql import convert_mongo_pipeline_to_sql

# 复杂管道：多次投影和分组
pipeline = [
    { "$match": { "status": "active", "age": { "$gte": 18 } } },
    { "$project": { "name": 1, "category": 1, "amount": 1 } },
    { "$group": { "_id": "$category", "totalAmount": { "$sum": "$amount" }, "count": { "$sum": 1 } } },
    { "$project": { "_id": 1, "totalAmount": 1, "count": 1 } },
    { "$match": { "totalAmount": { "$gte": 18 } } },
    { "$sort": { "totalAmount": -1 } },
    { "$limit": 5 }
]

sql, metadata = convert_mongo_pipeline_to_sql(pipeline, "transactions")
print(sql)
```

输出（自动使用 CTE 格式化）：
```sql
WITH cte_1 AS
  (SELECT t.category AS _id,
          SUM(t.amount) AS totalAmount,
          COUNT(*) AS COUNT
   FROM transactions AS t
   WHERE 1=1
     AND (t.status = 'active')
     AND (t.age >= 18)
   GROUP BY t.category),
     cte_2 AS
  (SELECT t._id AS _id,
          t.totalAmount AS totalAmount,
          t.count AS COUNT
   FROM cte_1 AS t)
SELECT *
FROM cte_2 AS t
WHERE 1=1
  AND (t.totalAmount >= 18)
ORDER BY t.totalAmount DESC
LIMIT 5
```

### 面向对象用法

```python
from mongo_2_sql import MongoToSQLConverter

converter = MongoToSQLConverter(dialect='sqlite')  # 当前主要支持 SQLite

pipeline = [
    { "$match": { "created_at": { "$gte": "2023-01-01" } } },
    { "$project": { 
        "name": 1, 
        "email": 1,
        "full_name": { "$concat": ["$first_name", " ", "$last_name"] }
    }},
    { "$sort": { "created_at": -1 } }
]

sql, metadata = converter.convert(pipeline, "users")
print(f"Generated SQL:\n{sql}")
print(f"\nMetadata: {metadata}")
```

### 处理元数据

```python
sql, metadata = convert_mongo_pipeline_to_sql(pipeline, "products")

print(f"SQL 方言: {metadata['dialect']}")
print(f"处理的阶段数: {metadata['stages_processed']}")
print(f"是否包含聚合: {metadata['has_aggregation']}")
print(f"SELECT 列: {metadata['select_columns']}")
print(f"WHERE 条件: {metadata['where_conditions']}")
print(f"警告信息: {metadata['warnings']}")
```

### 命令行用法

```bash
# 从文件转换
mongo2sql -f pipeline.json -c users

# 从命令行参数转换
mongo2sql -p '[{"$match": {"status": "active"}}]' -c users

# 从标准输入转换
echo '[{"$match": {"status": "active"}}]' | mongo2sql -c users

# 列出支持的阶段
mongo2sql --list-stages
```

## 支持的阶段

| MongoDB 阶段 | SQL 对应 | 描述 |
|-------------|---------|------|
| `$match` | `WHERE` | 过滤文档 |
| `$project` | `SELECT` | 选择/重塑字段 |
| `$group` | `GROUP BY` | 分组聚合 |
| `$sort` | `ORDER BY` | 排序结果 |
| `$limit` | `LIMIT` | 限制结果数量 |
| `$skip` | `OFFSET` | 跳过结果 |
| `$lookup` | `JOIN` | 关联查询 |
| `$unwind` | `UNNEST` | 展开数组 |

## 支持的操作符

### 比较操作符

- `$eq` - 等于 (=)
- `$ne` - 不等于 (!=)
- `$gt` - 大于 (>)
- `$gte` - 大于等于 (>=)
- `$lt` - 小于 (<)
- `$lte` - 小于等于 (<=)
- `$in` - 在数组中 (IN)
- `$nin` - 不在数组中 (NOT IN)

### 逻辑操作符

- `$and` - 逻辑与 (AND)
- `$or` - 逻辑或 (OR)
- `$not` - 逻辑非 (NOT)
- `$nor` - 逻辑或非 (NOR)

### 聚合函数

- `$sum` - 求和
- `$avg` - 平均值
- `$min` - 最小值
- `$max` - 最大值
- `$count` - 计数
- `$first` - 第一个值
- `$last` - 最后一个值

### 字符串函数

- `$concat` - 字符串连接
- `$toLower` - 转小写
- `$toUpper` - 转大写
- `$substr` - 子字符串

### 条件表达式

- `$cond` - 条件判断 (CASE WHEN)
- `$ifNull` - 空值处理 (COALESCE)

## 示例

### 示例 1: 简单查询

```python
pipeline = [
    { "$match": { "status": "active" } }
]
```

```sql
SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND (t.status = 'active')
```

### 示例 2: 比较操作符

```python
pipeline = [
    { "$match": { "age": { "$gte": 18, "$lt": 65 } } }
]
```

```sql
SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND ((t.age >= 18) AND (t.age < 65))
```

### 示例 3: 逻辑操作符

```python
pipeline = [
    { 
        "$match": { 
            "$or": [
                { "status": "active" },
                { "status": "pending" }
            ]
        } 
    }
]
```

```sql
SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND ((t.status = 'active') OR (t.status = 'pending'))
```

### 示例 4: 分组聚合

```python
pipeline = [
    { 
        "$group": { 
            "_id": "$category",
            "count": { "$sum": 1 },
            "avgPrice": { "$avg": "$price" }
        } 
    }
]
```

```sql
SELECT 
    t.category AS _id,
    COUNT(*) AS count,
    AVG(t.price) AS avgPrice
FROM products AS t
GROUP BY t.category
```

### 示例 5: 字段重命名

```python
pipeline = [
    { 
        "$project": { 
            "product_name": "$name",
            "total_value": { "$multiply": ["$price", "$quantity"] }
        } 
    }
]
```

```sql
SELECT 
    t.name AS product_name,
    (t.price * t.quantity) AS total_value
FROM products AS t
```

### 示例 6: 字符串连接

```python
pipeline = [
    { 
        "$project": { 
            "fullName": { "$concat": ["$firstName", " ", "$lastName"] }
        } 
    }
]
```

```sql
SELECT 
    CONCAT(
        t.firstName, 
        ' ', 
        t.lastName
    ) AS fullName
FROM users AS t
```

### 示例 7: JOIN 查询

```python
pipeline = [
    {
        "$lookup": {
            "from": "orders",
            "localField": "_id",
            "foreignField": "customerId",
            "as": "orders"
        }
    }
]
```

```sql
SELECT 
    *
FROM customers AS t
LEFT JOIN orders AS orders
    ON t._id = orders.customerId
```

## 高级特性

### 自定义阶段处理器

```python
from mongo_2_sql.core.stage_base import StageProcessor, RenderContext

class CustomStage(StageProcessor):
    def __init__(self):
        super().__init__('$custom')
    
    def process(self, stage_value, context: RenderContext):
        # 自定义处理逻辑
        pass
    
    def validate(self, stage_value):
        # 验证逻辑
        return True

# 注册自定义阶段
from mongo_2_sql.core.stage_loader import StageLoader
stage_loader = StageLoader()
stage_loader.register_stage(CustomStage())
```

### SQL 格式化选项

```python
from mongo_2_sql.utils import format_sql

sql = "select * from users where id = 1"
formatted_sql = format_sql(sql, uppercase_keywords=True)
print(formatted_sql)
```

### 错误处理最佳实践

```python
from mongo_2_sql import convert_mongo_pipeline_to_sql

try:
    sql, metadata = convert_mongo_pipeline_to_sql(pipeline, "users")
    if metadata['warnings']:
        print(f"警告: {metadata['warnings']}")
    print(sql)
except ValueError as e:
    print(f"管道格式错误: {e}")
except RuntimeError as e:
    print(f"转换错误: {e}")
```

## 最佳实践

### 1. 管道设计建议

- 尽量将 `$match` 阶段放在管道开头以提高性能
- 合理使用 `$project` 来减少不必要的字段传输
- 在 `$group` 之前使用 `$sort` 可以优化分组性能

### 2. 性能优化

```python
# 好的做法：过滤条件前置
pipeline = [
    { "$match": { "status": "active", "created_at": { "$gte": "2023-01-01" } } },
    { "$project": { "name": 1, "email": 1, "amount": 1 } },
    { "$group": { "_id": "$name", "total": { "$sum": "$amount" } } }
]

# 避免的做法：不必要的字段处理
pipeline = [
    { "$project": { "name": 1, "email": 1, "amount": 1, "unused_field": 1 } },
    { "$match": { "status": "active" } },  # 过滤太晚
    { "$group": { "_id": "$name", "total": { "$sum": "$amount" } } }
]
```

### 3. 调试技巧

```python
# 查看生成的元数据
sql, metadata = convert_mongo_pipeline_to_sql(pipeline, "users")

# 检查各个组件
print("SELECT 列:", metadata['select_columns'])
print("WHERE 条件:", metadata['where_conditions'])
print("GROUP BY 列:", metadata['group_by_columns'])
print("ORDER BY 列:", metadata['order_by_columns'])

# 使用 SQL 构建器进行精细控制
from mongo_2_sql.utils import SQLBuilder

builder = SQLBuilder()
sql = (builder
    .select('name', 'email')
    .from_table('users')
    .where("status = 'active'")
    .order_by('created_at', 'DESC')
    .limit(100)
    .build())
```

## 常见问题

### Q: 为什么生成的 SQL 中有 `1=1`？

A: 这是为了方便动态添加 WHERE 条件。您可以安全地忽略它或在后续处理中移除。

### Q: 如何处理复杂的嵌套查询？

A: 使用 CTE（公用表表达式）或者将复杂查询分解为多个简单步骤。

### Q: 支持哪些 SQL 方言？

A: 当前主要支持 SQLite 方言。PostgreSQL 和 MySQL 支持需要通过实现相应的渲染器来扩展。

### Q: 如何扩展支持新的 MongoDB 操作符？

A: 在对应的表达式解析器中添加新的操作符处理逻辑，然后注册到表达式注册表中。


## 贡献

欢迎贡献！请遵循以下步骤：

1. Fork 项目
2. 创建功能分支 (`git checkout -b feature/amazing-feature`)
3. 提交更改 (`git commit -m 'Add amazing feature'`)
4. 推送到分支 (`git push origin feature/amazing-feature`)
5. 创建 Pull Request

## 许可证

MIT License - 详见 [LICENSE](LICENSE) 文件

## 变更日志

### v0.1.4 (2025-02-13)

**新增功能:**
- CTE 统一处理框架，支持 $addFields、$project、$group 阶段切片
- ExpressionResolver 统一表达式解析器，避免 stage 间耦合
- BaseExpressionParser 基础解析器类，减少代码重复
- SQL 格式化输出（使用 sqlparse）
- Project 阶段字段重命名支持
- Stage 解耦架构设计

**问题修复:**
- 修复 `_id` 字段双引号问题
- 修复聚合后字段引用错误
- 修复模块导入路径问题

### v0.1.3 (2025-02-07)

**改进:**
- 完善文档和示例

## 致谢

感谢所有贡献者和用户的支持！

### 特别感谢

- 所有提交 Issue 和 Pull Request 的开发者
- 提供测试用例和反馈的用户