[AL-2290] Add result field to Task and get_result()

labelbox/schema/task.py

@@ -1,4 +1,5 @@
 import logging
+import requests
 import time
 from typing import TYPE_CHECKING, Optional
 
@@ -31,6 +32,7 @@ class Task(DbObject):
     name = Field.String("name")
     status = Field.String("status")
     completion_percentage = Field.Float("completion_percentage")
+    result = Field.String("result")
     _user: Optional["User"] = None
 
     # Relationships
@@ -65,3 +67,13 @@ def wait_till_done(self, timeout_seconds=300) -> None:
             timeout_seconds -= check_frequency
             time.sleep(sleep_time_seconds)
             self.refresh()
+
+    def errors(self):
+        """ Downloads the result file from Task
+        """
+        if self.status == "FAILED" and self.result:
+            response = requests.get(self.result)
+            response.raise_for_status()
+            data = response.json()
+            return data.get('error')
+        return None

mypy.ini

@@ -11,4 +11,4 @@ ignore_missing_imports = True
 ignore_errors = True
 
 [mypy-labelbox]
-ignore_errors = True
+ignore_errors = True

tests/integration/test_task.py

@@ -0,0 +1,39 @@
+from labelbox import DataRow
+from labelbox.schema.data_row_metadata import DataRowMetadataField
+
+EMBEDDING_SCHEMA_ID = "ckpyije740000yxdk81pbgjdc"
+
+
+def test_task_errors(dataset, image_url):
+    client = dataset.client
+    embeddings = [0.0] * 128
+    task = dataset.create_data_rows([
+        {
+            DataRow.row_data:
+                image_url,
+            DataRow.metadata_fields: [
+                DataRowMetadataField(schema_id=EMBEDDING_SCHEMA_ID,
+                                     value=embeddings),
+                DataRowMetadataField(schema_id=EMBEDDING_SCHEMA_ID,
+                                     value=embeddings)
+            ]
+        },
+    ])
+    assert task in client.get_user().created_tasks()
+    task.wait_till_done()
+    assert task.status == "FAILED"
+    assert task.errors() is not None
+    assert 'message' in task.errors()
+
+
+def test_task_success(dataset, image_url):
+    client = dataset.client
+    task = dataset.create_data_rows([
+        {
+            DataRow.row_data: image_url,
+        },
+    ])
+    assert task in client.get_user().created_tasks()
+    task.wait_till_done()
+    assert task.status == "COMPLETE"
+    assert task.errors() is None