如何通过Python调用C++程序（mongocxx驱动查询MongoDB）并返回结果集

我来帮你搞定这个需求——从Python调用C++程序，用mongocxx驱动查询MongoDB再把结果返回给Python。这里有两种实用的实现方案，你可以根据场景选择：

方案一：通过标准输出(stdout)传递结果（最简单） #

这种方式不需要处理复杂的跨语言接口，C++程序把查询结果转成JSON输出到stdout，Python读取stdout内容再解析即可。

1. 完善C++查询代码 #

先把你没写完的C++代码补全，实现查询并输出JSON格式的结果：

#include <bsoncxx/builder/stream/document.hpp>
#include <bsoncxx/json.hpp>
#include <mongocxx/client.hpp>
#include <mongocxx/instance.hpp>
#include <mongocxx/uri.hpp>
#include <iostream>
#include <vector>
#include <string>

int main() {
    try {
        // 初始化mongocxx实例，必须放在最前面
        mongocxx::instance inst{};
        // 替换成你的MongoDB连接URI
        mongocxx::uri uri("mongodb://localhost:27017");
        mongocxx::client conn(uri);

        // 替换成你的数据库和集合名称
        auto db = conn["your_database"];
        auto coll = db["your_collection"];

        // 构建查询条件，这里示例查询所有文档，你可以按需修改
        bsoncxx::builder::stream::document filter_builder;
        auto filter = filter_builder << bsoncxx::builder::stream::finalize;

        // 执行查询获取游标
        auto cursor = coll.find(filter);

        // 将每个BSON文档转成JSON字符串，逐行输出到stdout
        for (auto&& doc : cursor) {
            std::cout << bsoncxx::to_json(doc) << std::endl;
        }
    } catch (const std::exception& e) {
        // 把错误信息输出到stderr，方便Python捕获
        std::cerr << "C++程序执行错误: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

2. 编译C++程序 #

用g++编译时需要链接mongocxx和bsoncxx库，命令示例：

g++ -o mongo_query mongo_query.cpp -lmongocxx -lbsoncxx

如果你的库安装在非标准路径，需要加上-L/usr/local/lib（替换成你的库路径）和-I/usr/local/include/mongocxx/v_noabi -I/usr/local/include/bsoncxx/v_noabi（替换成你的头文件路径）。

3. Python调用并解析结果 #

用Python的subprocess模块调用编译后的可执行文件，读取输出并解析JSON：

import subprocess
import json

def fetch_mongo_data():
    try:
        # 调用C++可执行文件，捕获标准输出和错误
        proc = subprocess.run(
            ["./mongo_query"],
            capture_output=True,
            text=True,
            check=True
        )
        # 按行分割输出，每一行是一个JSON文档
        json_lines = proc.stdout.strip().split('\n')
        # 解析每个JSON字符串为Python字典
        return [json.loads(line) for line in json_lines if line]
    except subprocess.CalledProcessError as e:
        print(f"C++程序执行失败: {e.stderr}")
        return []
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
        return []

if __name__ == "__main__":
    data = fetch_mongo_data()
    print("查询到的MongoDB数据：")
    for doc in data:
        print(doc)

方案二：编译成动态链接库，用ctypes直接调用（更高效） #

如果追求更高的性能，不想通过stdout传递数据，可以把C代码编译成动态库，用Python的ctypes模块直接调用C函数。

1. 编写C++动态库代码 #

注意要提供C兼容的接口（用extern "C"包裹），还要处理内存释放避免泄漏：

#include <bsoncxx/builder/stream/document.hpp>
#include <bsoncxx/json.hpp>
#include <mongocxx/client.hpp>
#include <mongocxx/instance.hpp>
#include <mongocxx/uri.hpp>
#include <vector>
#include <string>
#include <cstring>

// 定义C兼容的结构体，用于传递结果
typedef struct {
    const char** json_docs;
    size_t count;
} MongoQueryResult;

extern "C" {
    // 查询MongoDB的函数
    MongoQueryResult query_mongodb() {
        static std::vector<std::string> result_docs;
        result_docs.clear();

        try {
            mongocxx::instance inst{};
            mongocxx::uri uri("mongodb://localhost:27017");
            mongocxx::client conn(uri);

            auto db = conn["your_database"];
            auto coll = db["your_collection"];

            auto filter = bsoncxx::builder::stream::document{} << bsoncxx::builder::stream::finalize;
            auto cursor = coll.find(filter);

            for (auto&& doc : cursor) {
                result_docs.push_back(bsoncxx::to_json(doc));
            }
        } catch (const std::exception& e) {
            // 把错误信息加入结果
            result_docs.push_back(std::string("Error: ") + e.what());
        }

        // 分配C风格的字符串数组
        const char** c_docs = new const char*[result_docs.size()];
        for (size_t i = 0; i < result_docs.size(); ++i) {
            c_docs[i] = strdup(result_docs[i].c_str());
        }

        MongoQueryResult result;
        result.json_docs = c_docs;
        result.count = result_docs.size();
        return result;
    }

    // 释放结果内存的函数，必须调用避免泄漏
    void free_mongo_result(MongoQueryResult* result) {
        if (result->json_docs) {
            for (size_t i = 0; i < result->count; ++i) {
                free((void*)result->json_docs[i]);
            }
            delete[] result->json_docs;
            result->json_docs = nullptr;
            result->count = 0;
        }
    }
}

2. 编译动态库 #

g++ -shared -fPIC -o libmongoquery.so mongo_query_lib.cpp -lmongocxx -lbsoncxx

3. Python用ctypes调用动态库 #

import ctypes
import json

# 加载动态库
mongo_lib = ctypes.CDLL("./libmongoquery.so")

# 定义结果结构体类型
class MongoQueryResult(ctypes.Structure):
    _fields_ = [("json_docs", ctypes.POINTER(ctypes.c_char_p)),
                ("count", ctypes.c_size_t)]

# 声明函数的返回值和参数类型
mongo_lib.query_mongodb.restype = MongoQueryResult
mongo_lib.free_mongo_result.argtypes = [ctypes.POINTER(MongoQueryResult)]

def get_mongo_data():
    result = mongo_lib.query_mongodb()
    data = []
    try:
        for i in range(result.count):
            json_str = result.json_docs[i].decode('utf-8')
            try:
                data.append(json.loads(json_str))
            except json.JSONDecodeError:
                # 如果是错误信息，直接添加
                data.append({"error": json_str})
    finally:
        # 必须调用释放函数，避免内存泄漏
        mongo_lib.free_mongo_result(ctypes.byref(result))
    return data

if __name__ == "__main__":
    mongo_data = get_mongo_data()
    print("查询到的数据：")
    for doc in mongo_data:
        print(doc)

内容的提问来源于stack exchange，提问作者anubhav231989