Allow mango filters in _changes API

tammoippen · djc · commit 026992fbc686 · 2017-04-25T11:08:50.000+02:00
If `filter==‘selector’` then there should be another kwarg `_selector`
with a `dict` containing the selector / mango filter.
diff --git a/couchdb/client.py b/couchdb/client.py
@@ -1186,7 +1186,12 @@ def update_doc(self, name, docid=None, **options):
         return headers, body
 
     def _changes(self, **opts):
-        _, _, data = self.resource.get('_changes', **opts)
+        # use streaming `get` and `post` methods
+        if opts.get('filter') == '_selector':
+            selector = opts.pop('_selector', None)
+            _, _, data = self.resource.post('_changes', selector, **opts)
+        else:
+            _, _, data = self.resource.get('_changes', **opts)
         lines = data.iterchunks()
         for ln in lines:
             if not ln: # skip heartbeats
@@ -1205,7 +1210,12 @@ def changes(self, **opts):
         """
         if opts.get('feed') == 'continuous':
             return self._changes(**opts)
-        _, _, data = self.resource.get_json('_changes', **opts)
+
+        if opts.get('filter') == '_selector':
+            selector = opts.pop('_selector', None)
+            _, _, data = self.resource.post_json('_changes', selector, **opts)
+        else:
+            _, _, data = self.resource.get_json('_changes', **opts)
         return data
 
 
diff --git a/couchdb/tests/client.py b/couchdb/tests/client.py
@@ -570,6 +570,28 @@ def test_changes_conn_usable(self):
         # in a good state from the previous request.
         self.assertTrue(self.db.info()['doc_count'] == 0)
 
+    def test_changes_conn_usable_selector(self):
+        if self.server.version_info()[0] < 2:
+            return
+        # Consume a changes feed to get a used connection in the pool.
+        list(self.db.changes(feed='continuous',
+                             filter='_selector',
+                             timeout=0,
+                             _selector={'selector': {}}))
+        # Try using the connection again to make sure the connection was left
+        # in a good state from the previous request.
+        self.assertTrue(self.db.info()['doc_count'] == 0)
+
+    def test_changes_usable_selector(self):
+        if self.server.version_info()[0] < 2:
+            return
+        # Consume a changes feed to get a used connection in the pool.
+        list(self.db.changes(filter='_selector',
+                             _selector={'selector': {}}))
+        # Try using the connection again to make sure the connection was left
+        # in a good state from the previous request.
+        self.assertTrue(self.db.info()['doc_count'] == 0)
+
     def test_changes_heartbeat(self):
         def wakeup():
             time.sleep(.3)
