mongo-2-sql 0.1.1

A library to convert MongoDB aggregation pipelines to SQL parser

Mongo to SQL

Mongo to SQL 是一个将 MongoDB 聚合管道转换为 SQL 查询语句的工具。

功能特性

• 阶段基础方法支持: 支持 $match, $project, $group, $sort, $limit, $lookup, $unwind 等常用阶段
• 丰富的操作符: 支持比较操作符、逻辑操作符、聚合函数、条件表达式、字符串函数等
• 可扩展架构: 模块化的设计，易于添加新的阶段处理器和自定义函数
• 多方言支持: 主要支持 SQLite，可扩展支持 PostgreSQL、MySQL 等 SQL 方言
• 类型安全: 完整的类型注解支持，提供良好的 IDE 提示
• 格式化输出: 自动生成格式化的 SQL，提高可读性
• 错误处理: 完善的错误检查和友好的错误提示
• 性能优化: 针对不同 SQL 方言的特定优化

安装

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 阶段

快速开始

基本用法

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)

输出：

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

面向对象用法

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}")

处理元数据

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']}")

命令行用法

# 从文件转换
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: 简单查询

pipeline = [
    { "$match": { "status": "active" } }
]

SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND (t.status = 'active')

示例 2: 比较操作符

pipeline = [
    { "$match": { "age": { "$gte": 18, "$lt": 65 } } }
]

SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND ((t.age >= 18) AND (t.age < 65))

示例 3: 逻辑操作符

pipeline = [
    { 
        "$match": { 
            "$or": [
                { "status": "active" },
                { "status": "pending" }
            ]
        } 
    }
]

SELECT 
    *
FROM users AS t 
WHERE 1=1
 AND ((t.status = 'active') OR (t.status = 'pending'))

示例 4: 分组聚合

pipeline = [
    { 
        "$group": { 
            "_id": "$category",
            "count": { "$sum": 1 },
            "avgPrice": { "$avg": "$price" }
        } 
    }
]

SELECT 
    t.category AS _id,
    COUNT(*) AS count,
    AVG(t.price) AS avgPrice
FROM products AS t
GROUP BY t.category

示例 5: 字符串连接

pipeline = [
    { 
        "$project": { 
            "fullName": { "$concat": ["$firstName", " ", "$lastName"] }
        } 
    }
]

SELECT 
    CONCAT(
        t.firstName, 
        ' ', 
        t.lastName
    ) AS fullName
FROM users AS t

示例 6: JOIN 查询

pipeline = [
    {
        "$lookup": {
            "from": "orders",
            "localField": "_id",
            "foreignField": "customerId",
            "as": "orders"
        }
    }
]

SELECT 
    *
FROM customers AS t
LEFT JOIN orders AS orders
    ON t._id = orders.customerId

项目结构

mongo_2_sql/
├── core/                   # 核心模块
│   ├── ast_base.py        # AST 基础节点
│   ├── expr_base.py       # 表达式系统
│   ├── expr_registry.py   # 表达式注册表
│   ├── stage_base.py      # 阶段处理器基类
│   ├── stage_loader.py    # 阶段加载器
│   └── pipeline_runtime.py # 管道运行时
├── stages/                 # 阶段处理器
│   ├── match/             # $match 阶段
│   ├── project/           # $project 阶段
│   ├── group/             # $group 阶段
│   ├── sort/              # $sort 阶段
│   ├── limit/             # $limit/$skip 阶段
│   ├── lookup/            # $lookup 阶段
│   └── unwind/            # $unwind 阶段
├── renderer/               # SQL 渲染器
│   └── sqlite.py          # SQLite 渲染器
├── converter.py            # 主要转换入口
├── main.py                 # 命令行入口
├── utils.py                # 工具函数
└── examples.py             # 使用示例

高级特性

自定义阶段处理器

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 格式化选项

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)

错误处理最佳实践

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. 性能优化

# 好的做法：过滤条件前置
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. 调试技巧

# 查看生成的元数据
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 文件

变更日志

v0.1.0 (2025-02-07)

新特性:

• 初始版本发布
• 支持基本的 MongoDB 聚合阶段转换
• 实现 SQLite 方言支持（PostgreSQL、MySQL 需要扩展）
• 添加命令行工具

改进:

• 优化 SQL 格式化输出
• 增强错误处理和验证
• 完善文档和示例

修复:

• 修复 $project 阶段表达式处理问题
• 修正 WHERE 条件分组逻辑
• 解决引号处理不一致问题
• 补充指令操作

致谢

感谢所有贡献者和用户的支持！

特别感谢

• 所有提交 Issue 和 Pull Request 的开发者
• 提供测试用例和反馈的用户