[ACTIONS] MongoDb - Aggregation, Update Many Documents, Find a Document

components/mongodb/actions/execute-aggregation/execute-aggregation.mjs

@@ -0,0 +1,64 @@
+import app from "../../mongodb.app.mjs";
+import utils from "../../common/utils.mjs";
+
+export default {
+  key: "mongodb-execute-aggregation",
+  name: "Execute Aggregation",
+  description: "Execute an aggregation pipeline on a MongoDB collection. [See the documentation](https://www.mongodb.com/docs/drivers/node/current/fundamentals/aggregation/)",
+  version: "0.0.1",
+  type: "action",
+  props: {
+    app,
+    database: {
+      propDefinition: [
+        app,
+        "database",
+      ],
+    },
+    collectionName: {
+      propDefinition: [
+        app,
+        "collection",
+        ({ database }) => ({
+          database,
+        }),
+      ],
+    },
+    pipeline: {
+      type: "string[]",
+      label: "Pipeline",
+      description: "The aggregation pipeline to execute where each row represents a stage as a JSON string. Eg. `[ { \"$match\": { \"categories\": \"Bakery\" } }, { \"$group\": { \"_id\": \"$stars\", \"count\": { \"$sum\": 1 } } } ]`",
+    },
+  },
+  methods: {
+    async excecuteAggregation({
+      database, collectionName, pipeline,
+    } = {}) {
+      const { app } = this;
+      const client = await app.getClient();
+      const collection = app.getCollection(client, database, collectionName);
+      const cursor = collection.aggregate(pipeline);
+      const result = await utils.iterate(cursor);
+      await client.close();
+      return result;
+    },
+  },
+  async run({ $ }) {
+    const {
+      excecuteAggregation,
+      database,
+      collectionName,
+      pipeline,
+    } = this;
+
+    const response = await excecuteAggregation({
+      database,
+      collectionName,
+      pipeline: utils.parseArray(pipeline),
+    });
+
+    $.export("$summary", `Successfully executed aggregation pipeline on collection with \`${response.length}\` document(s).`);
+
+    return response;
+  },
+};

components/mongodb/actions/find-document/find-document.mjs

@@ -0,0 +1,75 @@
+import app from "../../mongodb.app.mjs";
+import utils from "../../common/utils.mjs";
+
+export default {
+  key: "mongodb-find-document",
+  name: "Find Document",
+  description: "Finds a document by a query filter. [See the documentation](https://docs.mongodb.com/manual/reference/method/db.collection.find/)",
+  version: "0.0.1",
+  type: "action",
+  props: {
+    app,
+    database: {
+      propDefinition: [
+        app,
+        "database",
+      ],
+    },
+    collectionName: {
+      propDefinition: [
+        app,
+        "collection",
+        ({ database }) => ({
+          database,
+        }),
+      ],
+    },
+    filter: {
+      type: "string",
+      label: "Filter",
+      description: "JSON string to use as a filter. Eg. `{ \"name\": \"Twitter\" }`",
+    },
+    options: {
+      type: "string",
+      label: "Options",
+      description: "JSON string to use as options. Eg. `{ \"projection\": { \"_id\": 0, \"title\": 1 } }`",
+      optional: true,
+    },
+  },
+  methods: {
+    async findDocument({
+      database, collectionName, filter, options,
+    } = {}) {
+      const { app } = this;
+      const client = await app.getClient();
+      const collection = app.getCollection(client, database, collectionName);
+      const result = await collection.findOne(filter, options);
+      await client.close();
+      return result;
+    },
+  },
+  async run({ $ }) {
+    const {
+      findDocument,
+      database,
+      collectionName,
+      filter,
+      options,
+    } = this;
+
+    const response = await findDocument({
+      database,
+      collectionName,
+      filter: utils.valueToObject(filter),
+      options: utils.valueToObject(options),
+    });
+
+    if (!response) {
+      $.export("$summary", "Document not found in the collection.");
+      return;
+    }
+
+    $.export("$summary", "Successfully found a document in the collection.");
+    return response;
+  },
+};

components/mongodb/actions/update-documents/update-documents.mjs

@@ -0,0 +1,69 @@
+import app from "../../mongodb.app.mjs";
+import utils from "../../common/utils.mjs";
+
+export default {
+  key: "mongodb-update-documents",
+  name: "Update Documents",
+  description: "Updates many documents by query filter. [See the documentation](https://www.mongodb.com/docs/drivers/node/current/usage-examples/updateMany/)",
+  version: "0.0.1",
+  type: "action",
+  props: {
+    app,
+    database: {
+      propDefinition: [
+        app,
+        "database",
+      ],
+    },
+    collectionName: {
+      propDefinition: [
+        app,
+        "collection",
+        ({ database }) => ({
+          database,
+        }),
+      ],
+    },
+    filter: {
+      type: "string",
+      label: "Filter",
+      description: "JSON string to use as a filter. Eg. `{ \"rated\": \"G\" }`",
+    },
+    data: {
+      label: "Data To Update",
+      type: "string",
+      description: "JSON data to use as the update. Eg. `{ \"$set\": { \"rating\": \"Everyone\" } }`",
+    },
+  },
+  methods: {
+    async updateDocuments({
+      database, collectionName, filter, data,
+    } = {}) {
+      const { app } = this;
+      const client = await app.getClient();
+      const collection = app.getCollection(client, database, collectionName);
+      const result = await collection.updateMany(filter, data);
+      await client.close();
+      return result;
+    },
+  },
+  async run({ $ }) {
+    const {
+      updateDocuments,
+      database,
+      collectionName,
+      filter,
+      data,
+    } = this;
+
+    const response = await updateDocuments({
+      database,
+      collectionName,
+      filter: utils.valueToObject(filter),
+      data: utils.valueToObject(data),
+    });
+
+    $.export("$summary", `Successfully updated \`${response.modifiedCount}\` document(s) in the collection.`);
+    return response;
+  },
+};

components/mongodb/common/utils.mjs

@@ -0,0 +1,63 @@
+import { ConfigurationError } from "@pipedream/platform";
+
+function isJson(value) {
+  if (typeof(value) !== "string") {
+    return false;
+  }
+
+  try {
+    JSON.parse(value);
+  } catch (e) {
+    return false;
+  }
+
+  return true;
+}
+
+function valueToObject(value) {
+  if (value === undefined || typeof(value) === "object") {
+    return value;
+  }
+
+  if (!isJson(value)) {
+    throw new ConfigurationError(`Make sure the value contains a valid JSON string: ${value}`);
+  }
+  return JSON.parse(value);
+}
+
+function parseArray(value) {
+  try {
+    if (!value) {
+      return [];
+    }
+
+    if (Array.isArray(value)) {
+      return value;
+    }
+
+    const parsedValue = JSON.parse(value);
+
+    if (!Array.isArray(parsedValue)) {
+      throw new Error("Not an array");
+    }
+
+    return parsedValue;
+
+  } catch (e) {
+    throw new ConfigurationError("Make sure the value contains a valid JSON array");
+  }
+}
+
+async function iterate(iterations) {
+  const items = [];
+  for await (const item of iterations) {
+    items.push(item);
+  }
+  return items;
+}
+
+export default {
+  valueToObject,
+  parseArray: (value) => parseArray(value).map(valueToObject),
+  iterate,
+};

components/mongodb/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@pipedream/mongodb",
-  "version": "0.0.8",
+  "version": "0.1.0",
   "description": "Pipedream MongoDB Components",
   "main": "mongodb.app.mjs",
   "keywords": [
@@ -14,7 +14,7 @@
     "access": "public"
   },
   "dependencies": {
-    "@pipedream/platform": "^1.2.0",
+    "@pipedream/platform": "^1.6.5",
     "mongodb": "^4.6.0"
   }
 }