RESTClient: rewrite client.py to handle cluster & async

cloudify/manager.py

@@ -8,7 +8,8 @@
 from cloudify.state import ctx, workflow_ctx, NotInContext
 from cloudify.exceptions import (HttpException,
                                  NonRecoverableError)
-from cloudify.cluster import CloudifyClusterClient
+from cloudify_rest_client.client import CloudifyClient
+from cloudify_async_client.client import AsyncCloudifyClient
 
 
 class NodeInstance(object):
@@ -123,7 +124,7 @@ def system_properties(self):
         return self._system_properties
 
 
-def get_rest_client(tenant=None, api_token=None):
+def get_rest_client(tenant=None, api_token=None, async_client=False):
     """
     :param tenant: optional tenant name to connect as
     :param api_token: optional api_token to authenticate with (instead of
@@ -150,7 +151,11 @@ def get_rest_client(tenant=None, api_token=None):
     else:
         token = utils.get_rest_token()
 
-    return CloudifyClusterClient(
+    client_cls = CloudifyClient
+    if async_client:
+        client_cls = AsyncCloudifyClient
+
+    return client_cls(
         headers=headers,
         host=utils.get_manager_rest_service_host(),
         port=utils.get_manager_rest_service_port(),
@@ -159,7 +164,8 @@ def get_rest_client(tenant=None, api_token=None):
         protocol=utils.get_manager_rest_service_protocol(),
         cert=utils.get_local_rest_certificate(),
         kerberos_env=utils.get_kerberos_indication(
-            os.environ.get(constants.KERBEROS_ENV_KEY))
+            os.environ.get(constants.KERBEROS_ENV_KEY)),
+        retries=30,
     )
 
 

cloudify_async_client/__init__.py

@@ -1 +1,3 @@
-from cloudify_async_client.client import CloudifyAsyncClient  # noqa
+from cloudify_async_client.client import AsyncCloudifyClient
+
+__all__ = ['AsyncCloudifyClient']

cloudify_async_client/audit_log.py

@@ -1,4 +1,3 @@
-from cloudify_async_client import CloudifyAsyncClient
 from cloudify_rest_client.audit_log import AuditLogClient
 
 
@@ -14,20 +13,10 @@ async def stream(self, timeout=None, **kwargs):
         :return:        ``ListResponse`` with of ``AuditLog`` items and
                         response metadata.
         """
-        client = await self.async_client()
-        response = await client.get('audit/stream',
-                                    params=kwargs,
-                                    timeout=timeout)
-        return response
-
-    async def async_client(self):
-        headers = self.api.headers.copy()
-        headers.update({'Content-type': 'text/event-stream'})
-        client = CloudifyAsyncClient(
-            host=self.api.host,
-            port=self.api.port,
-            protocol=self.api.protocol,
-            cert=self.api.cert,
-            headers=headers,
+        response = await self.api.get(
+            '/audit/stream',
+            params=kwargs,
+            timeout=timeout,
+            stream=True,
         )
-        return client
+        return response