PYTHON-207 Update M/R API for server 1.8.0

The map_reduce method now requires an 'out' parameter. Clients that were already passing out="<collection name>" will continue to function correctly. Also added merge_output and reduce_output options which will only work correctly with server >= 1.7.4. This change also adds an inline_map_reduce method that also only works with server >= 1.7.4.

Author: behackett

pymongo/collection.py

@@ -21,7 +21,7 @@
 from pymongo import (helpers,
  message)
 from pymongo.cursor import Cursor
-from pymongo.errors import InvalidName
+from pymongo.errors import InvalidName, InvalidOperation
 
 _ZERO = "\x00\x00\x00\x00"
 
@@ -914,7 +914,8 @@ def distinct(self, key):
  """
  return self.find().distinct(key)
 
- def map_reduce(self, map, reduce, full_response=False, **kwargs):
+ def map_reduce(self, map, reduce, out, merge_output=False,
+ reduce_output=False, full_response=False, **kwargs):
  """Perform a map/reduce operation on this collection.
 
  If `full_response` is ``False`` (default) returns a
@@ -925,6 +926,15 @@ def map_reduce(self, map, reduce, full_response=False, **kwargs):
  :Parameters:
  - `map`: map function (as a JavaScript string)
  - `reduce`: reduce function (as a JavaScript string)
+ - `out` (required): output collection name
+ - `merge_output` (optional): Merge output into `out`. If the same
+ key exists in both the result set and the existing output collection,
+ the new key will overwrite the existing key
+ - `reduce_output` (optional): If documents exist for a given key
+ in the result set and in the existing output collection, then a
+ reduce operation (using the specified reduce function) will be
+ performed on the two values and the result will be written to
+ the output collection
  - `full_response` (optional): if ``True``, return full response to
  this command - otherwise just return the result collection
  - `**kwargs` (optional): additional arguments to the
@@ -943,11 +953,62 @@ def map_reduce(self, map, reduce, full_response=False, **kwargs):
 
  .. mongodoc:: mapreduce
  """
+ if not isinstance(out, basestring):
+ raise TypeError("'out' must be an instance of basestring")
+
+ if merge_output and reduce_output:
+ raise InvalidOperation("Can't do both merge and re-reduce of output.")
+
+ if merge_output:
+ out_conf = {"merge": out}
+ elif reduce_output:
+ out_conf = {"reduce": out}
+ else:
+ out_conf = out
+
+ response = self.__database.command("mapreduce", self.__name,
+ map=map, reduce=reduce,
+ out=out_conf, **kwargs)
+ if full_response:
+ return response
+ else:
+ return self.__database[response["result"]]
+
+ def inline_map_reduce(self, map, reduce, full_response=False, **kwargs):
+ """Perform an inline map/reduce operation on this collection.
+
+ Perform the map/reduce operation on the server in RAM. A result
+ collection is not created. The result set is returned as a list
+ of documents.
+
+ If `full_response` is ``False`` (default) returns the
+ result documents in a list. Otherwise, returns the full
+ response from the server to the `map reduce command`_.
+
+ :Parameters:
+ - `map`: map function (as a JavaScript string)
+ - `reduce`: reduce function (as a JavaScript string)
+ - `full_response` (optional): if ``True``, return full response to
+ this command - otherwise just return the result collection
+ - `**kwargs` (optional): additional arguments to the
+ `map reduce command`_ may be passed as keyword arguments to this
+ helper method, e.g.::
+
+ >>> db.test.map_reduce(map, reduce, limit=2)
+
+ .. note:: Requires server version **>= 1.7.4**
+
+ .. versionadded:: 1.10
+ """
+
  response = self.__database.command("mapreduce", self.__name,
- map=map, reduce=reduce, **kwargs)
+ map=map, reduce=reduce,
+ out={"inline": 1}, **kwargs)
+
  if full_response:
  return response
- return self.__database[response["result"]]
+ else:
+ return response.get("results")
 
  def find_and_modify(self, query={}, update=None, upsert=False, **kwargs):
  """Update and return an object.

test/test_collection.py

@@ -1034,6 +1034,31 @@ def test_map_reduce(self):
  self.assertEqual(2, result.find_one({"_id": "dog"})["value"])
  self.assertEqual(1, result.find_one({"_id": "mouse"})["value"])
 
+ if version.at_least(self.db.connection, (1, 7, 4)):
+ db.test.insert({"id": 5, "tags": ["hampster"]})
+ result = db.test.map_reduce(map, reduce, out='mrunittests')
+ self.assertEqual(1, result.find_one({"_id": "hampster"})["value"])
+ db.test.remove({"id": 5})
+ result = db.test.map_reduce(map, reduce,
+ out='mrunittests', merge_output=True)
+ self.assertEqual(3, result.find_one({"_id": "cat"})["value"])
+ self.assertEqual(1, result.find_one({"_id": "hampster"})["value"])
+
+ result = db.test.map_reduce(map, reduce,
+ out='mrunittests', reduce_output=True)
+ self.assertEqual(6, result.find_one({"_id": "cat"})["value"])
+ self.assertEqual(4, result.find_one({"_id": "dog"})["value"])
+ self.assertEqual(2, result.find_one({"_id": "mouse"})["value"])
+ self.assertEqual(1, result.find_one({"_id": "hampster"})["value"])
+
+ self.assertRaises(InvalidOperation,
+ db.test.map_reduce,
+ map,
+ reduce,
+ out='mrunittests',
+ merge_output=True,
+ reduce_output=True)
+
  full_result = db.test.map_reduce(map, reduce,
  out='mrunittests', full_response=True)
  self.assertEqual(6, full_result["counts"]["emit"])
@@ -1043,6 +1068,16 @@ def test_map_reduce(self):
  self.assertEqual(1, result.find_one({"_id": "dog"})["value"])
  self.assertEqual(None, result.find_one({"_id": "mouse"}))
 
+ if version.at_least(self.db.connection, (1, 7, 4)):
+ result = db.test.inline_map_reduce(map, reduce)
+ self.assertTrue(isinstance(result, list))
+ self.assertEqual(3, len(result))
+ self.assertTrue(result[1]["_id"] in ("cat", "dog", "mouse"))
+
+ full_result = db.test.inline_map_reduce(map, reduce,
+ full_response=True)
+ self.assertEqual(6, full_result["counts"]["emit"])
+
  def test_messages_with_unicode_collection_names(self):
  db = self.db