fixes qubole#144 in qubole/qds-sdk-py

When creating clusters with the v1.3 of the API we still use the v1.2 endpoint
during cluster creation. The _parse_create_update method accepts the
api-version only to distinguish the request parameters. But the create, update
and clone methods still use the default configuration of Qubole.agent(..) which
is set to v1.2 unless all the actions in the session are configure(..)`ed with
the v1.3 of the API.

If you want to perform some of the unsupported actions with v1.3 of the
cluster API but fall back to v1.2 for commands APIs we explicitly switch the
version in the create(..)/update(..)/clone(..) calls which thereby alter the
version for that API call alone and leave the cached_agent as is for the v1.2
APIs.

qds_sdk/cluster.py

@@ -24,6 +24,7 @@ class Cluster(Resource):
     """
 
     rest_entity_path = "clusters"
+    api_version = "v1.2"
 
     @classmethod
     def _parse_list(cls, args):
@@ -408,29 +409,38 @@ def _parse_create_update(cls, args, action, api_version):
         return arguments
 
     @classmethod
-    def create(cls, cluster_info):
+    def create(cls, cluster_info, version=None):
         """
         Create a new cluster using information provided in `cluster_info`.
+
+        Optionally provide the version (eg: v1.3) to use the new version of the
+        API. If None we default to v1.2
         """
-        conn = Qubole.agent()
+        conn = Qubole.agent(version=version)
         return conn.post(cls.rest_entity_path, data=cluster_info)
 
     @classmethod
-    def update(cls, cluster_id_label, cluster_info):
+    def update(cls, cluster_id_label, cluster_info, version=None):
         """
         Update the cluster with id/label `cluster_id_label` using information provided in
         `cluster_info`.
+
+        Optionally provide the version (eg: v1.3) to use the new version of the
+        API. If None we default to v1.2
         """
-        conn = Qubole.agent()
+        conn = Qubole.agent(version=version)
         return conn.put(cls.element_path(cluster_id_label), data=cluster_info)
 
     @classmethod
-    def clone(cls, cluster_id_label, cluster_info):
+    def clone(cls, cluster_id_label, cluster_info, version=None):
         """
         Update the cluster with id/label `cluster_id_label` using information provided in
         `cluster_info`.
+
+        Optionally provide the version (eg: v1.3) to use the new version of the
+        API. If None we default to v1.2
         """
-        conn = Qubole.agent()
+        conn = Qubole.agent(version=version)
         return conn.post(cls.element_path(cluster_id_label) + '/clone', data=cluster_info)
 
     @classmethod
@@ -1155,7 +1165,6 @@ def minimal_payload(self):
         creating or updating a cluster.
         """
         payload_dict = self.__dict__
-        payload_dict.pop("api_version", None)
         return _make_minimal(payload_dict)
 
 

qds_sdk/connection.py

@@ -31,9 +31,9 @@ def init_poolmanager(self, connections, maxsize,
 
 class Connection:
 
-    def __init__(self, auth, base_url, skip_ssl_cert_check, reuse=True):
+    def __init__(self, auth, rest_url, skip_ssl_cert_check, reuse=True):
         self.auth = auth
-        self.base_url = base_url
+        self.rest_url = rest_url
         self.skip_ssl_cert_check = skip_ssl_cert_check
         self._headers = {'User-Agent': 'qds-sdk-py-%s' % pkg_resources.get_distribution("qds-sdk").version,
                          'Content-Type': 'application/json'}
@@ -61,7 +61,7 @@ def delete(self, path, data=None):
         return self._api_call("DELETE", path, data)
 
     def _api_call_raw(self, req_type, path, data=None, params=None):
-        url = self.base_url.rstrip('/') + '/' + path
+        url = self.rest_url.rstrip('/') + '/' + path
 
         if self.reuse:
             x = self.session

qds_sdk/qubole.py

@@ -25,7 +25,8 @@ class Qubole:
 
     _auth = None
     api_token = None
-    base_url = None
+    baseurl = None
+    version = None
     poll_interval = None
     skip_ssl_cert_check = None
 
@@ -41,13 +42,14 @@ def configure(cls, api_token,
 
             `api_url`: the base URL for QDS API. configurable for testing only
 
-            `version`: QDS REST api version
+            `version`: QDS REST api version. Will be used throughout unless overridden in Qubole.agent(..)
 
             `poll_interval`: interval in secs when polling QDS for events
         """
         cls._auth = QuboleAuth(api_token)
         cls.api_token = api_token
-        cls.base_url = api_url.rstrip('/') + '/' + version
+        cls.version = version
+        cls.baseurl = api_url
         if poll_interval < Qubole.MIN_POLL_INTERVAL:
             log.warn("Poll interval cannot be less than %s seconds. Setting it to %s seconds.\n" % (Qubole.MIN_POLL_INTERVAL, Qubole.MIN_POLL_INTERVAL))
             cls.poll_interval = Qubole.MIN_POLL_INTERVAL
@@ -58,15 +60,30 @@ def configure(cls, api_token,
     cached_agent = None
 
     @classmethod
-    def agent(cls):
+    def agent(cls, version=None):
         """
         Returns:
            a connection object to make REST calls to QDS
+
+           optionally override the `version` of the REST endpoint for advanced
+           features available only in the newer version of the API available
+           for certain resource end points eg: /v1.3/cluster. When version is
+           None we default to v1.2
         """
+        reuse_cached_agent = True
+        if version:
+          log.debug("api version changed to %s" % version)
+          cls.rest_url = '/'.join([cls.baseurl.rstrip('/'), version])
+          reuse_cached_agent = False
+        else:
+          cls.rest_url = '/'.join([cls.baseurl.rstrip('/'), cls.version])
         if cls.api_token is None:
             raise ConfigError("No API Token specified - please supply one via Qubole.configure()")
 
+        if not reuse_cached_agent:
+          uncached_agent = Connection(cls._auth, cls.rest_url, cls.skip_ssl_cert_check)
+          return uncached_agent
         if cls.cached_agent is None:
-            cls.cached_agent = Connection(cls._auth, cls.base_url, cls.skip_ssl_cert_check)
+          cls.cached_agent = Connection(cls._auth, cls.rest_url, cls.skip_ssl_cert_check)
 
         return cls.cached_agent