add tailable option to find() to use a tailable cursor

Author: Mike Dirolf

pymongo/collection.py

@@ -305,7 +305,7 @@ def _fields_list_to_dict(self, fields):
         return as_dict
 
     def find(self, spec=None, fields=None, skip=0, limit=0,
-             slave_okay=None, timeout=True, snapshot=False,
+             slave_okay=None, timeout=True, snapshot=False, tailable=False,
              _sock=None, _must_use_master=False):
         """Query the database.
 
@@ -342,9 +342,15 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
           - `snapshot` (optional): if True, snapshot mode will be used for this
             query. Snapshot mode assures no duplicates are returned, or objects
             missed, which were present at both the start and end of the query's
-            execution. For details, see the wiki_
-
-        .. _wiki: http://www.mongodb.org/display/DOCS/How+to+do+Snapshotting+in+the+Mongo+Database
+            execution. For details, see the `wiki
+            <http://www.mongodb.org/display/DOCS/How+to+do+Snapshotting+in+the+Mongo+Database>`_.
+          - `tailable` (optional): the result of this find call will be a
+            tailable cursor - tailable cursors aren't closed when the last data
+            is retrieved but are kept open and the cursors location marks the
+            final document's position. if more data is received iteration of
+            the cursor will continue from the last document received. For
+            details, see the `wiki
+            <http://www.mongodb.org/display/DOCS/Tailable+Cursors>`_.
         """
         if spec is None:
             spec = SON()
@@ -370,14 +376,17 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
             raise TypeError("timeout must be an instance of bool")
         if not isinstance(snapshot, types.BooleanType):
             raise TypeError("snapshot must be an instance of bool")
+        if not isinstance(tailable, types.BooleanType):
+            raise TypeError("tailable must be an instance of bool")
 
         if fields is not None:
             if not fields:
                 fields = ["_id"]
             fields = self._fields_list_to_dict(fields)
 
-        return Cursor(self, spec, fields, skip, limit, slave_okay, timeout, snapshot,
-                      _sock=_sock, _must_use_master=_must_use_master)
+        return Cursor(self, spec, fields, skip, limit, slave_okay, timeout,
+                      tailable, snapshot, _sock=_sock,
+                      _must_use_master=_must_use_master)
 
     def count(self):
         """Get the number of documents in this collection.

pymongo/cursor.py

@@ -37,7 +37,8 @@ class Cursor(object):
     """
 
     def __init__(self, collection, spec, fields, skip, limit, slave_okay,
-                 timeout, snapshot=False, _sock=None, _must_use_master=False):
+                 timeout, tailable, snapshot=False,
+                 _sock=None, _must_use_master=False):
         """Create a new cursor.
 
         Should not be called directly by application developers.
@@ -49,6 +50,7 @@ def __init__(self, collection, spec, fields, skip, limit, slave_okay,
         self.__limit = limit
         self.__slave_okay = slave_okay
         self.__timeout = timeout
+        self.__tailable = tailable
         self.__snapshot = snapshot
         self.__ordering = None
         self.__explain = False
@@ -99,7 +101,7 @@ def clone(self):
         """
         copy = Cursor(self.__collection, self.__spec, self.__fields,
                       self.__skip, self.__limit, self.__slave_okay,
-                      self.__timeout, self.__snapshot)
+                      self.__timeout, self.__tailable, self.__snapshot)
         copy.__ordering = self.__ordering
         copy.__explain = self.__explain
         copy.__hint = self.__hint
@@ -141,6 +143,8 @@ def __query_options(self):
         """Get the 4 byte query options string to use for this query.
         """
         options = 0
+        if self.__tailable:
+            options |= _QUERY_OPTIONS["tailable_cursor"]
         if self.__slave_okay:
             options |= _QUERY_OPTIONS["slave_okay"]
         if not self.__timeout:
@@ -381,7 +385,11 @@ def __send_message(self, operation, message):
 
         response = pymongo.Connection._unpack_response(response, self.__id)
         self.__id = response["cursor_id"]
-        assert response["starting_from"] == self.__retrieved
+
+        # starting from doesn't get set on getmore's for tailable cursors
+        if not self.__tailable:
+            assert response["starting_from"] == self.__retrieved
+
         self.__retrieved += response["number_returned"]
         self.__data = response["data"]
 

test/test_cursor.py

@@ -555,6 +555,36 @@ def set_coll():
 
         self.assertRaises(AttributeError, set_coll)
 
+    def test_tailable(self):
+        db = self.db
+        db.drop_collection("test")
+
+
+        cursor = db.test.find(tailable=True)
+
+        db.test.insert({"x": 1})
+        count = 0
+        for doc in cursor:
+            count += 1
+            self.assertEqual(1, doc["x"])
+        self.assertEqual(1, count)
+
+        db.test.insert({"x": 2})
+        count = 0
+        for doc in cursor:
+            count += 1
+            self.assertEqual(2, doc["x"])
+        self.assertEqual(1, count)
+
+        db.test.insert({"x": 3})
+        count = 0
+        for doc in cursor:
+            count += 1
+            self.assertEqual(3, doc["x"])
+        self.assertEqual(1, count)
+
+        self.assertEqual(3, db.test.count())
+
 
 if __name__ == "__main__":
     unittest.main()