mongo-2-sql 0.1.5

A library to convert MongoDB aggregation pipelines to SQL parser

Mongo to SQL

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

项目仓库

• GitHub: baobingbo/mongo-2-sql

版本更新 v0.1.5

新增功能

1. $addFields 阶段表达式与字面量语义

• $addFields 使用独立的 AddFieldsExpressionParser，不再复用 $project 的解析逻辑，避免跨阶段耦合
• $addFields 支持三类值：
  • 字面量值: 如 {"score": 1}, { "flag": true }, { "text": "hello" }，会直接转换为 1 AS score、TRUE AS flag、'hello' AS text
  • 字段引用: 如 { "cat": "$category" }，会转换为 t.category AS cat
  • 表达式: 如 { "total": { "$add": [1, "$price"] } } 或 { "isBig": { "$gt": ["$amount", 100] } }
• 仅 $project 阶段中，{"field": 1} / {"field": 0} 仍然表示“包含/排除字段”；其他阶段（包括 $addFields）中，1 始终被视为字面量数值 1

功能特性

• 阶段基础方法支持: 支持 $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

CTE 处理（复杂聚合管道）

当管道包含多次 $project、$group 或 $addFields 阶段时，系统会自动使用 CTE 方式处理：

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 格式化）：

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

面向对象用法

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": { 
            "product_name": "$name",
            "total_value": { "$multiply": ["$price", "$quantity"] }
        } 
    }
]

SELECT 
    t.name AS product_name,
    (t.price * t.quantity) AS total_value
FROM products AS t

示例 6: 字符串连接

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

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

示例 7: 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

高级特性

自定义阶段处理器

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.5 (2025-02-23)

新增功能:

• addFields阶段独立解析器（AddFieldsExpressionParser），与 project 解耦
• 支持字面量值、字段引用、表达式三类语义
• 统一数值 1 为字面量（仅 $project 阶段保留包含/排除语义）

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 的开发者
• 提供测试用例和反馈的用户