MongoDB 将相关 collection 项计数与其他 collection 结果合并

Question

我是 mongodb 的新手，正在尝试了解如何有效地查询 collection 中的每个项目。

我有 projects collection & tasks collections

//projects
{
   _id: ObjectId(),
   name: String
}
//tasks
{
   _id: ObjectId(),
   projectId: ObjectId(), //reference project id
   completed: Bool
}

我想获取所有项目，然后计算每个项目的 completed 和 incomplete 任务

db.projects.find({})...
//perhaps something similar in output
[
 {
   _id: ObjectId(), //projectId
   name: String
   completed: Number,
   incomplete: Number
 }
]

我正在使用猫鼬作为 ORM。我不知道这在猫鼬甚至本机 mongodb 查询中是否可行。感谢任何帮助。谢谢！

Answer 1

当您需要对 MongoDB 中的事物进行分组或计数时，您通常需要使用 aggregation framework。以下是如何计算 shell 中的数据：

db.tasks.aggregate([ {$group: {
  _id: {projectID: "$projectID", completed: "$completed"},
  count: {$sum: 1}
  }});

这将为项目中的每个任务 return 两个文档 - 一个包含已完成任务的计数，另一个包含尚未完成的任务。

我从未使用过 Mongoose，但现在您可以开始学习了:)

Answer 2

无论你怎么看，只要你有像这样的规范化关系，那么你就需要两个查询来获得包含来自 "tasks" collection 的详细信息的结果，并填写详情来自 "projects" collection。 MongoDB 不以任何方式使用连接，mongoose 也不例外。 Mongoose 确实提供 .populate()，但这只是本质上运行另一个查询的便利魔法，并将结果合并到引用的字段值上。

所以在这种情况下，您可能最终会考虑将项目信息嵌入到任务中。当然会有重复，但是使用单数 collection.

可以使查询模式更加简单

保持 collections 与参考模型分开，您基本上有两种方法。但首先您可以使用 aggregate 以获得更符合您实际要求的结果：

      Task.aggregate(
        [
          { "$group": {
            "_id": "$projectId",
            "completed": {
              "$sum": {
                "$cond": [ "$completed", 1, 0 ]
              }
            },
            "incomplete": {
              "$sum": {
                "$cond": [ "$completed", 0, 1 ]
              }
            }
          }}
        ],
        function(err,results) {

        }
    );

这仅仅使用了一个$group pipeline in order to accumulate on the values of "projectid" witin the "tasks" collection. In order to count the values for "completed" and "incomplete" we use the $cond operator which is a ternary to decide which value to pass to $sum。由于这里的第一个或 "if" 条件是一个布尔值评估，那么现有的布尔值 "complete" 字段就可以了，传递 where true 到 "then" 或 "else" 传递第三个参数。

这些结果没问题，但它们不包含来自 "project" collection 的任何关于收集到的“_id”值的信息。使输出看起来像这样的一种方法是从返回的 "results" object:

的聚合结果回调中调用 .populate() 的模型形式

    Project.populate(results,{ "path": "_id" },callback);

在这种形式中，.populate() 调用采用 object 或数据数组作为第一个参数，第二个参数是人口的选项文档，此处的必填字段是 "path"。这将处理模型中的任何项目和 "populate"，并将这些 object 插入回调中的结果数据中。

作为完整示例清单：

var async = require('async'),
    mongoose = require('mongoose'),
    Schema = mongoose.Schema;

var projectSchema = new Schema({
  "name": String
});

var taskSchema = new Schema({
  "projectId": { "type": Schema.Types.ObjectId, "ref": "Project" },
  "completed": { "type": Boolean, "default": false }
});

var Project = mongoose.model( "Project", projectSchema );
var Task = mongoose.model( "Task", taskSchema );

mongoose.connect('mongodb://localhost/test');

async.waterfall(
  [
    function(callback) {
      async.each([Project,Task],function(model,callback) {
        model.remove({},callback);
      },
      function(err) {
        callback(err);
      });
    },

    function(callback) {
      Project.create({ "name": "Project1" },callback);
    },

    function(project,callback) {
      Project.create({ "name": "Project2" },callback);
    },

    function(project,callback) {
      Task.create({ "projectId": project },callback);
    },

    function(task,callback) {
      Task.aggregate(
        [
          { "$group": {
            "_id": "$projectId",
            "completed": {
              "$sum": {
                "$cond": [ "$completed", 1, 0 ]
              }
            },
            "incomplete": {
              "$sum": {
                "$cond": [ "$completed", 0, 1 ]
              }
            }
          }}
        ],
        function(err,results) {
          if (err) callback(err);
          Project.populate(results,{ "path": "_id" },callback);
        }
      );
    }
  ],
  function(err,results) {
    if (err) throw err;
    console.log( JSON.stringify( results, undefined, 4 ));
    process.exit();
  }
);

这将给出如下结果：

[
    {
        "_id": {
            "_id": "54beef3178ef08ca249b98ef",
            "name": "Project2",
            "__v": 0
        },
        "completed": 0,
        "incomplete": 1
    }
]

所以 .populate() 适用于这种聚合结果，甚至是另一个查询，并且通常应该适用于大多数用途。然而，清单中包含一个特定示例，其中创建了 "two" 个项目，但当然只有 "one" 个任务仅引用其中一个项目。

由于聚合在 "tasks" collection 上进行，因此它不知道那里未引用的任何 "project"。为了获得 "projects" 的完整列表以及计算出的总数，您需要在运行两个查询和 "merging" 结果中更加具体。

这基本上是一个关于不同键和数据的 "hash merge"，但是这个模块的好帮手是一个名为 nedb 的模块，它允许您以与 [= 更一致的方式应用逻辑100=] 查询和操作。

基本上，您想要一份来自 "projects" collection 的数据副本，并带有扩充字段，然后您想要 "merge" 或 .update() 该信息与聚合结果.再次作为完整清单进行演示：

var async = require('async'),
    mongoose = require('mongoose'),
    Schema = mongoose.Schema,
    DataStore = require('nedb'),
    db = new DataStore();


var projectSchema = new Schema({
  "name": String
});

var taskSchema = new Schema({
  "projectId": { "type": Schema.Types.ObjectId, "ref": "Project" },
  "completed": { "type": Boolean, "default": false }
});

var Project = mongoose.model( "Project", projectSchema );
var Task = mongoose.model( "Task", taskSchema );

mongoose.connect('mongodb://localhost/test');

async.waterfall(
  [
    function(callback) {
      async.each([Project,Task],function(model,callback) {
        model.remove({},callback);
      },
      function(err) {
        callback(err);
      });
    },

    function(callback) {
      Project.create({ "name": "Project1" },callback);
    },

    function(project,callback) {
      Project.create({ "name": "Project2" },callback);
    },

    function(project,callback) {
      Task.create({ "projectId": project },callback);
    },

    function(task,callback) {
      async.series(
        [

          function(callback) {
            Project.find({},function(err,projects) {
              async.eachLimit(projects,10,function(project,callback) {
                db.insert({
                  "projectId": project._id.toString(),
                  "name": project.name,
                  "completed": 0,
                  "incomplete": 0
                },callback);
              },callback);
            });
          },

          function(callback) {
            Task.aggregate(
              [
                { "$group": {
                  "_id": "$projectId",
                  "completed": {
                    "$sum": {
                      "$cond": [ "$completed", 1, 0 ]
                    }
                  },
                  "incomplete": {
                    "$sum": {
                      "$cond": [ "$completed", 0, 1 ]
                    }
                  }
                }}
              ],
              function(err,results) {
                async.eachLimit(results,10,function(result,callback) {
                  db.update(
                    { "projectId": result._id.toString() },
                    { "$set": {
                        "complete": result.complete,
                        "incomplete": result.incomplete
                      }
                    },
                    callback
                  );
                },callback);
              }
            );
          },

        ],

        function(err) {
          if (err) callback(err);
          db.find({},{ "_id": 0 },callback);
        }
      );
    }
  ],
  function(err,results) {
    if (err) throw err;
    console.log( JSON.stringify( results, undefined, 4 ));
    process.exit();
  }

结果在这里：

[
    {
        "projectId": "54beef4c23d4e4e0246379db",
        "name": "Project2",
        "completed": 0,
        "incomplete": 1
    },
    {
        "projectId": "54beef4c23d4e4e0246379da",
        "name": "Project1",
        "completed": 0,
        "incomplete": 0
    }
]

列出每个 "project" 的数据并包括与其相关的 "tasks" collection 的计算值。

因此，您可以采用几种方法。同样，您最终可能最好只是将 "tasks" 嵌入到 "project" 项目中，这又是一种简单的聚合方法。如果你打算嵌入任务信息，那么你也可以在 "project" object 上维护 "complete" 和 "incomplete" 的计数器，并在标记项目时简单地更新这些使用 $inc 运算符在任务数组中完成。

var taskSchema = new Schema({
  "completed": { "type": Boolean, "default": false }
});

var projectSchema = new Schema({
  "name": String,
  "completed": { "type": Number, "default": 0 },
  "incomplete": { "type": Number, "default": 0 }
  "tasks": [taskSchema]
});

var Project = mongoose.model( "Project", projectSchema );
// cheat for a model object with no collection
var Task = mongoose.model( "Task", taskSchema, undefined );

// Then in later code

// Adding a task
var task = new Task();
Project.update(
    { "task._id": { "$ne": task._id } },
    { 
        "$push": { "tasks": task },
        "$inc": {
            "completed": ( task.completed ) ? 1 : 0,
            "incomplete": ( !task.completed ) ? 1 : 0;
        }
    },
    callback
 );

// Removing a task
Project.update(
    { "task._id": task._id },
    { 
        "$pull": { "tasks": { "_id": task._id } },
        "$inc": {
            "completed": ( task.completed ) ? -1 : 0,
            "incomplete": ( !task.completed ) ? -1 : 0;
        }
    },
    callback
 );


 // Marking complete
Project.update(
    { "tasks": { "$elemMatch": { "_id": task._id, "completed": false } }},
    { 
        "$set": { "tasks.$.completed": true },
        "$inc": {
            "completed": 1,
            "incomplete": -1
        }
    },
    callback
);

您必须知道当前的任务状态，但计数器更新才能正常工作，但这很容易编写代码，您可能应该至少在 object 中传递这些详细信息到您的方法中。

就我个人而言，我会 re-model 使用后一种形式并执行此操作。您可以执行查询 "merging"，如此处的两个示例所示，但它当然是有代价的。

MongoDB 将相关 collection 项计数与其他 collection 结果合并

MongoDB merge related collection item count with other collection results

node.js

mongodb

mongoose

mongodb-query

aggregation-framework