fix mypy

Author: Matt Sokoloff

labelbox/schema/project.py

@@ -327,7 +327,7 @@ def _validate_datetime(string_date: str) -> bool:
                 try:
                     datetime.strptime(string_date, "%Y-%m-%d")
                 except ValueError:
-                    raise ValueError(f"""Incorrect format for: {string_date}. 
+                    raise ValueError(f"""Incorrect format for: {string_date}.
                     Format must be \"YYYY-MM-DD\"""")
             return True
 

labelbox/schema/task.py

@@ -1,8 +1,8 @@
 import logging
+import json
 import requests
 import time
-from typing import TYPE_CHECKING, Optional, Dict, Any
-from functools import lru_cache
+from typing import TYPE_CHECKING, TypeVar, Callable, Optional, Dict, Any, List
 
 from labelbox.exceptions import ResourceNotFoundError
 from labelbox.orm.db_object import DbObject
@@ -11,6 +11,11 @@
 if TYPE_CHECKING:
     from labelbox import User
 
+    def lru_cache() -> Callable[..., Callable[..., Dict[str, Any]]]:
+        pass
+else:
+    from functools import lru_cache
+
 logger = logging.getLogger(__name__)
 
 
@@ -73,33 +78,40 @@ def wait_till_done(self, timeout_seconds=300) -> None:
     def errors(self) -> Optional[Dict[str, Any]]:
         """ Downloads the result file from Task
         """
-        self.wait_till_done(timeout_seconds=600)
         if self.status == "FAILED":
-            data = self._fetch_remote(self.result_url)
-            if data:
-                return data.get('error', None)
-        elif self.status == "IN_PROGRESS":
-            raise Exception("Job state IN_PROGRESS. Result not available.")
+            result = self._fetch_remote_json()
+            return result['error']
         return None
 
     @property
-    def result(self) -> Dict[str, Any]:
+    def result(self) -> List[Dict[str, Any]]:
         """ Fetch the result for a task
         """
-        self.wait_till_done(timeout_seconds=600)
-        if self.status == "COMPLETE":
-            return self._fetch_remote(self.result_url)
-        elif self.status == "FAILED":
-            errors = self.errors
-            message = errors.get('message') or errors
-            raise Exception(f"Job failed. Errors : {message}")
+        if self.status == "FAILED":
+            raise ValueError(f"Job failed. Errors : {self.errors}")
         else:
-            raise Exception("Job state IN_PROGRESS. Result not available.")
+            result = self._fetch_remote_json()
+            return [{
+                'id': data_row['id'],
+                'external_id': data_row.get('externalId'),
+                'row_data': data_row['rowData']
+            } for data_row in result['createdDataRows']]
 
     @lru_cache()
-    def _fetch_remote(self, result_url) -> Dict[str, Any]:
+    def _fetch_remote_json(self) -> Dict[str, Any]:
         """ Function for fetching and caching the result data.
         """
-        response = requests.get(result_url)
+        if self.name != 'JSON Import':
+            raise ValueError(
+                "Task result is only supported for `JSON Import` tasks."
+                " Download task.result_url manually to access the result for other tasks."
+            )
+        self.wait_till_done(timeout_seconds=600)
+        if self.status == "IN_PROGRESS":
+            raise ValueError(
+                "Job status still in `IN_PROGRESS`. The result is not available. Increase timeout or contact support."
+            )
+
+        response = requests.get(self.result_url)
         response.raise_for_status()
         return response.json()

tests/data/metrics/confusion_matrix/t.py

@@ -557,7 +557,5 @@ def test_create_data_rows_result(client, dataset, image_url):
         },
     ])
     assert task.errors is None
-    assert 'createdDataRows' in task.result
-    results = task.result['createdDataRows']
-    for result in results:
+    for result in task.result:
         client.get_data_row(result['id'])

tests/integration/test_data_rows.py

@@ -1,6 +1,6 @@
 import pytest
 
-from labelbox import DataRow
+from labelbox import DataRow, Task
 from labelbox.schema.data_row_metadata import DataRowMetadataField
 
 EMBEDDING_SCHEMA_ID = "ckpyije740000yxdk81pbgjdc"
@@ -28,11 +28,10 @@ def test_task_errors(dataset, image_url):
     assert 'message' in task.errors
     with pytest.raises(Exception) as exc_info:
         task.result
-    assert str(exc_info.value).startswith(
-        "Job failed. Errors : Failed to validate the metadata")
+    assert str(exc_info.value).startswith("Job failed. Errors : {")
 
 
-def test_task_success(dataset, image_url):
+def test_task_success_json(dataset, image_url):
     client = dataset.client
     task = dataset.create_data_rows([
         {
@@ -44,3 +43,18 @@ def test_task_success(dataset, image_url):
     assert task.status == "COMPLETE"
     assert task.errors is None
     assert task.result is not None
+    assert len(task.result)
+
+
+def test_task_success_label_export(client, configured_project_with_label):
+    project, _, _, _ = configured_project_with_label
+    project.export_labels()
+    user = client.get_user()
+    task = None
+    for task in user.created_tasks():
+        if task.name != 'JSON Import':
+            break
+
+    with pytest.raises(ValueError) as exc_info:
+        task.result
+    assert str(exc_info.value).startswith("Task result is only supported for")