Add support for _scheduler endpoints (#379)

* Add support for _scheduler endpoints

* PR updates:

- add get_doc for replication docs by docid
- fix methods so they return the json, not the
  whole response

* Fetch features/metadata, update replication_state

* Copyrights, docstrings

* Copyright

* Fix TODO # Local Variables:

* Attempt to fix broken test

* Add new expected state for scheduler endpoint

* Add new expected state for scheduler endpoint

Author: tomblench

src/cloudant/client.py

@@ -108,6 +108,7 @@ def __init__(self, user, auth_token, admin_party=False, **kwargs):
             if (not user and not auth_token) and (parsed_url.username and parsed_url.password):
                 self._user = parsed_url.username
                 self._auth_token = parsed_url.password
+        self._features = None
 
         connect_to_couch = kwargs.get('connect', False)
         if connect_to_couch and self._DATABASE_CLASS == CouchDatabase:
@@ -122,6 +123,18 @@ def is_iam_authenticated(self):
         """
         return self._use_iam
 
+    def features(self):
+        """
+        lazy fetch and cache features
+        """
+        if self._features is None:
+            metadata = self.metadata()
+            if "features" in metadata:
+                self._features = metadata["features"]
+            else:
+                self._features = []
+        return self._features
+
     def connect(self):
         """
         Starts up an authentication session for the client using cookie
@@ -324,6 +337,16 @@ def db_updates(self, raw_data=False, **kwargs):
         """
         return Feed(self, raw_data, **kwargs)
 
+    def metadata(self):
+        """
+        Retrieves the remote server metadata dictionary.
+
+        :returns: Dictionary containing server metadata details
+        """
+        resp = self.r_session.get(self.server_url)
+        resp.raise_for_status()
+        return resp.json()
+
     def keys(self, remote=False):
         """
         Returns the database names for this client. Default is

src/cloudant/replicator.py

@@ -18,8 +18,11 @@
 
 import uuid
 
+from requests.exceptions import HTTPError
+
 from .error import CloudantReplicatorException, CloudantClientException
 from .document import Document
+from .scheduler import Scheduler
 
 class Replicator(object):
     """
@@ -34,6 +37,7 @@ class Replicator(object):
 
     def __init__(self, client):
         repl_db = '_replicator'
+        self.client = client
         try:
             self.database = client[repl_db]
         except Exception:
@@ -133,12 +137,20 @@ def replication_state(self, repl_id):
 
         :returns: Replication state as a ``str``
         """
-        try:
-            repl_doc = self.database[repl_id]
-        except KeyError:
-            raise CloudantReplicatorException(404, repl_id)
-        repl_doc.fetch()
-        return repl_doc.get('_replication_state')
+        if "scheduler" in self.client.features():
+            try:
+                repl_doc = Scheduler(self.client).get_doc(repl_id)
+            except HTTPError as err:
+                raise CloudantReplicatorException(err.response.status_code, repl_id)
+            state = repl_doc['state']
+        else:
+            try:
+                repl_doc = self.database[repl_id]
+            except KeyError:
+                raise CloudantReplicatorException(404, repl_id)
+            repl_doc.fetch()
+            state = repl_doc.get('_replication_state')
+        return state
 
     def follow_replication(self, repl_id):
         """
@@ -161,12 +173,19 @@ def update_state():
             """
             Retrieves the replication state.
             """
-            try:
-                arepl_doc = self.database[repl_id]
-                arepl_doc.fetch()
-                return arepl_doc, arepl_doc.get('_replication_state')
-            except KeyError:
-                return None, None
+            if "scheduler" in self.client.features():
+                try:
+                    arepl_doc = Scheduler(self.client).get_doc(repl_id)
+                    return arepl_doc, arepl_doc['state']
+                except HTTPError:
+                    return None, None
+            else:
+                try:
+                    arepl_doc = self.database[repl_id]
+                    arepl_doc.fetch()
+                    return arepl_doc, arepl_doc.get('_replication_state')
+                except KeyError:
+                    return None, None
 
         while True:
             # Make sure we fetch the state up front, just in case it moves

src/cloudant/scheduler.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+# Copyright (C) 2018 IBM Corp. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+API module for interacting with scheduler endpoints
+"""
+
+class Scheduler(object):
+    """
+    API for retrieving scheduler jobs and documents.
+
+    :param client: Client instance used by the database.  Can either be a
+        ``CouchDB`` or ``Cloudant`` client instance.
+    """
+
+    def __init__(self, client):
+        self._client = client
+        self._r_session = client.r_session
+        self._scheduler = '/'.join([self._client.server_url, '_scheduler'])
+
+    def list_docs(self, limit=None, skip=None):
+        """
+        Lists replication documents. Includes information
+        about all the documents, even in completed and failed
+        states. For each document it returns the document ID, the
+        database, the replication ID, source and target, and other
+        information.
+
+        :param limit: How many results to return.
+        :param skip: How many result to skip starting at the beginning, if ordered by document ID.
+        """
+        params = dict()
+        if limit != None:
+            params["limit"] = limit
+        if skip != None:
+            params["skip"] = skip
+        resp = self._r_session.get('/'.join([self._scheduler, 'docs']), params=params)
+        resp.raise_for_status()
+        return resp.json()
+
+    def get_doc(self, doc_id):
+        """
+        Get replication document state for a given replication document ID.
+        """
+        resp = self._r_session.get('/'.join([self._scheduler, 'docs', '_replicator', doc_id]))
+        resp.raise_for_status()
+        return resp.json()
+
+
+    def list_jobs(self, limit=None, skip=None):
+        """
+        Lists replication jobs. Includes replications created via
+        /_replicate endpoint as well as those created from replication
+        documents. Does not include replications which have completed
+        or have failed to start because replication documents were
+        malformed. Each job description will include source and target
+        information, replication id, a history of recent event, and a
+        few other things.
+
+        :param limit: How many results to return.
+        :param skip: How many result to skip starting at the beginning, if ordered by document ID.
+        """
+        params = dict()
+        if limit != None:
+            params["limit"] = limit
+        if skip != None:
+            params["skip"] = skip
+        resp = self._r_session.get('/'.join([self._scheduler, 'jobs']), params=params)
+        resp.raise_for_status()
+        return resp.json()

tests/unit/replicator_tests.py

@@ -317,7 +317,7 @@ def test_retrieve_replication_state(self):
         )
         self.replication_ids.append(repl_id)
         repl_state = None
-        valid_states = ['completed', 'error', 'triggered', None]
+        valid_states = ['completed', 'error', 'triggered', 'running', None]
         finished = False
         for _ in range(300):
             repl_state = self.replicator.replication_state(repl_id)
@@ -402,11 +402,15 @@ def test_follow_replication(self):
             repl_id
         )
         self.replication_ids.append(repl_id)
-        valid_states = ('completed', 'error', 'triggered', None)
+        valid_states = ('completed', 'error', 'triggered', 'running', None)
         repl_states = []
+        if 'scheduler' in self.client.features():
+            state_key = 'state'
+        else:
+            state_key = '_replication_state'        
         for doc in self.replicator.follow_replication(repl_id):
-            self.assertIn(doc.get('_replication_state'), valid_states)
-            repl_states.append(doc.get('_replication_state'))
+            self.assertIn(doc.get(state_key), valid_states)
+            repl_states.append(doc.get(state_key))
         self.assertTrue(len(repl_states) > 0)
         self.assertEqual(repl_states[-1], 'completed')
         self.assertNotIn('error', repl_states)