Merge pull request #32 from sypht-team/add-async-workfows

add run_workflow_async method, blackify

Author: duaneallam

sypht/client.py

@@ -75,9 +75,9 @@ def _create_session(self):
         return requests.Session()
 
     def _authenticate_v2(self, endpoint, client_id, client_secret, audience):
-        basic_auth_slug = b64encode((client_id + ":" + client_secret).encode("utf-8")).decode(
-            "utf-8"
-        )
+        basic_auth_slug = b64encode(
+            (client_id + ":" + client_secret).encode("utf-8")
+        ).decode("utf-8")
         result = self.requests.post(
             endpoint,
             headers={
@@ -95,7 +95,9 @@ def _authenticate_v2(self, endpoint, client_id, client_secret, audience):
         return result["access_token"], result["expires_in"]
 
     def _authenticate_v1(self, endpoint, client_id, client_secret, audience):
-        endpoint = endpoint or os.environ.get("SYPHT_AUTH_ENDPOINT", SYPHT_LEGACY_AUTH_ENDPOINT)
+        endpoint = endpoint or os.environ.get(
+            "SYPHT_AUTH_ENDPOINT", SYPHT_LEGACY_AUTH_ENDPOINT
+        )
         result = self.requests.post(
             endpoint,
             data={
@@ -107,7 +109,9 @@ def _authenticate_v1(self, endpoint, client_id, client_secret, audience):
         ).json()
 
         if result.get("error_description"):
-            raise Exception("Authentication failed: {}".format(result["error_description"]))
+            raise Exception(
+                "Authentication failed: {}".format(result["error_description"])
+            )
 
         return result["access_token"], result["expires_in"]
 
@@ -214,13 +218,36 @@ def upload(
 
         if "fileId" not in result:
             raise Exception(
-                "Upload failed with response: {}".format("\n" + json.dumps(result, indent=2))
+                "Upload failed with response: {}".format(
+                    "\n" + json.dumps(result, indent=2)
+                )
             )
 
         return result["fileId"]
 
     def run_workflow(self, workflow, inputs, step=None, endpoint=None, headers=None):
-        endpoint = urljoin(endpoint or self.base_endpoint, f"workflows/{workflow}/invoke")
+        endpoint = urljoin(
+            endpoint or self.base_endpoint, f"workflows/{workflow}/invoke"
+        )
+        headers = headers or {}
+        headers = self._get_headers(**headers)
+        return self._parse_response(
+            self.requests.post(
+                endpoint,
+                data=json.dumps(
+                    {
+                        "step_id": step,
+                        "inputs": inputs,
+                    }
+                ),
+                headers=headers,
+            )
+        )
+
+    def run_workflow_async(
+        self, workflow, inputs, step=None, endpoint=None, headers=None
+    ):
+        endpoint = urljoin(endpoint or self.base_endpoint, f"workflows/{workflow}/jobs")
         headers = headers or {}
         headers = self._get_headers(**headers)
         return self._parse_response(
@@ -332,7 +359,9 @@ def get_file(self, file_id, endpoint=None, headers=None):
         return self._parse_response(self.requests.get(endpoint, headers=headers))
 
     def get_file_data(self, file_id, endpoint=None, headers=None):
-        endpoint = urljoin(endpoint or self.base_endpoint, f"app/docs/{file_id}/download")
+        endpoint = urljoin(
+            endpoint or self.base_endpoint, f"app/docs/{file_id}/download"
+        )
         headers = headers or {}
         headers = self._get_headers(**headers)
         response = self.requests.get(endpoint, headers=headers)
@@ -401,9 +430,13 @@ def get_annotations_for_docs(self, doc_ids, endpoint=None):
         headers = self._get_headers()
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
-        return self._parse_response(self.requests.post(endpoint, data=body, headers=headers))
+        return self._parse_response(
+            self.requests.post(endpoint, data=body, headers=headers)
+        )
 
-    def set_company_annotations(self, doc_id, annotations, company_id=None, endpoint=None):
+    def set_company_annotations(
+        self, doc_id, annotations, company_id=None, endpoint=None
+    ):
         data = {
             "origin": "external",
             "fields": [
@@ -481,7 +514,9 @@ def set_files_for_tag(self, tag, file_ids, company_id=None, endpoint=None):
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
         return self._parse_response(
-            self.requests.put(endpoint, data=json.dumps({"docs": file_ids}), headers=headers)
+            self.requests.put(
+                endpoint, data=json.dumps({"docs": file_ids}), headers=headers
+            )
         )
 
     def add_files_to_tag(self, tag, file_ids, company_id=None, endpoint=None):
@@ -494,7 +529,9 @@ def add_files_to_tag(self, tag, file_ids, company_id=None, endpoint=None):
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
         return self._parse_response(
-            self.requests.patch(endpoint, data=json.dumps({"docs": file_ids}), headers=headers)
+            self.requests.patch(
+                endpoint, data=json.dumps({"docs": file_ids}), headers=headers
+            )
         )
 
     def remove_file_from_tag(self, file_id, tag, company_id=None, endpoint=None):
@@ -529,7 +566,9 @@ def set_tags_for_file(self, file_id, tags, company_id=None, endpoint=None):
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
         return self._parse_response(
-            self.requests.put(endpoint, data=json.dumps({"tags": tags}), headers=headers)
+            self.requests.put(
+                endpoint, data=json.dumps({"tags": tags}), headers=headers
+            )
         )
 
     def add_tags_to_file(self, file_id, tags, company_id=None, endpoint=None):
@@ -542,7 +581,9 @@ def add_tags_to_file(self, file_id, tags, company_id=None, endpoint=None):
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
         return self._parse_response(
-            self.requests.patch(endpoint, data=json.dumps({"tags": tags}), headers=headers)
+            self.requests.patch(
+                endpoint, data=json.dumps({"tags": tags}), headers=headers
+            )
         )
 
     def get_entity(self, entity_id, entity_type, company_id=None, endpoint=None):
@@ -586,7 +627,9 @@ def get_many_entities(self, entity_type, entities, company_id=None, endpoint=Non
             self.requests.post(endpoint, data=json.dumps(entities), headers=headers)
         )
 
-    def list_entities(self, entity_type, company_id=None, page=None, limit=None, endpoint=None):
+    def list_entities(
+        self, entity_type, company_id=None, page=None, limit=None, endpoint=None
+    ):
         """Get list of entity_ids by pagination."""
         company_id = company_id or self.company_id
         entity_type = quote_plus(entity_type)
@@ -602,9 +645,13 @@ def list_entities(self, entity_type, company_id=None, page=None, limit=None, end
             params["page"] = page
         if limit:
             params["limit"] = int(limit)
-        return self._parse_response(self.requests.get(endpoint, headers=headers, params=params))
+        return self._parse_response(
+            self.requests.get(endpoint, headers=headers, params=params)
+        )
 
-    def get_all_entity_ids(self, entity_type, verbose=True, company_id=None, endpoint=None):
+    def get_all_entity_ids(
+        self, entity_type, verbose=True, company_id=None, endpoint=None
+    ):
         """Get all entity_ids for specified entity_type.
 
         Returns list of objects if verbose (by default):
@@ -687,7 +734,9 @@ def set_many_entities(
         for batch in _iter_chunked_sequence(entities, batch_size):
             responses.append(
                 self._parse_response(
-                    self.requests.post(endpoint, data=json.dumps(batch), headers=headers)
+                    self.requests.post(
+                        endpoint, data=json.dumps(batch), headers=headers
+                    )
                 )
             )
         return responses
@@ -720,7 +769,9 @@ def update_specification(self, specification, endpoint=None):
         headers["Accept"] = "application/json"
         headers["Content-Type"] = "application/json"
         return self._parse_response(
-            self.requests.post(endpoint, data=json.dumps(specification), headers=headers)
+            self.requests.post(
+                endpoint, data=json.dumps(specification), headers=headers
+            )
         )
 
     def submit_task(

tests/tests_client.py

@@ -20,7 +20,9 @@ class DataExtraction(unittest.TestCase):
     def setUp(self):
         warnings.simplefilter("ignore", category=ResourceWarning)
 
-        self.sypht_client = SyphtClient(os.environ["CLIENT_ID"], os.environ["CLIENT_SECRET"])
+        self.sypht_client = SyphtClient(
+            os.environ["CLIENT_ID"], os.environ["CLIENT_SECRET"]
+        )
 
     def test_with_wrong_fieldset(self):
         with self.assertRaises(Exception) as context: