Add set_log_upload_error for DOs (#5448)

* add set_log_status for DOs

* add log status to do service

* update naming to be log error instead of status

fiftyone/factory/repos/delegated_operation.py

@@ -117,6 +117,12 @@ def set_label(
         """Sets the label for the delegated operation."""
         raise NotImplementedError("subclass must implement set_label()")
 
+    def set_log_upload_error(
+        self, _id: ObjectId, log_upload_error: str
+    ) -> DelegatedOperationDocument:
+        """Sets the log upload error for the delegated operation."""
+        raise NotImplementedError("subclass must implement set_log_upload_error()")
+
     def get(self, _id: ObjectId) -> DelegatedOperationDocument:
         """Get an operation by id."""
         raise NotImplementedError("subclass must implement get()")
@@ -167,6 +173,13 @@ def _create_indexes(self):
                 )
             )
 
+        if "dataset_id_1" not in index_names:
+            indices_to_create.append(
+                IndexModel(
+                    [("dataset_id", pymongo.ASCENDING)], name="dataset_id_1"
+                )
+            )
+
         if indices_to_create:
             self._collection.create_indexes(indices_to_create)
 
@@ -239,6 +252,16 @@ def set_label(
         )
         return DelegatedOperationDocument().from_pymongo(doc)
 
+    def set_log_upload_error(
+        self, _id: ObjectId, log_upload_error: str
+    ) -> DelegatedOperationDocument:
+        doc = self._collection.find_one_and_update(
+            filter={"_id": _id},
+            update={"$set": {"log_upload_error": log_upload_error}},
+            return_document=pymongo.ReturnDocument.AFTER,
+        )
+        return DelegatedOperationDocument().from_pymongo(doc)
+
     def update_run_state(
         self,
         _id: ObjectId,

fiftyone/factory/repos/delegated_operation_doc.py

@@ -56,6 +56,7 @@ def __init__(
         self.id = None
         self._doc = None
         self.metadata = None
+        self.log_upload_error = None
 
     def from_pymongo(self, doc: dict):
         # required fields
@@ -72,6 +73,7 @@ def from_pymongo(self, doc: dict):
         self.pinned = doc.get("pinned", None)
         self.dataset_id = doc.get("dataset_id", None)
         self.run_link = doc.get("run_link", None)
+        self.log_upload_error = doc.get("log_upload_error", None)
         self.metadata = doc.get("metadata", None)
         self.label = doc.get("label", None)
         self.updated_at = doc.get("updated_at", None)

fiftyone/operators/delegated.py

@@ -252,6 +252,19 @@ def set_label(self, doc_id, label):
         """
         return self._repo.set_label(_id=doc_id, label=label)
 
+    def set_log_upload_error(self, doc_id, log_upload_error):
+        """Sets the log upload error for the given delegated operation.
+
+        Args:
+            doc_id: the ID of the delegated operation
+            log upload error: the error message if we failed to upload
+            logs for the given delegated operation.
+
+        Returns:
+            a :class:`fiftyone.factory.repos.DelegatedOperationDocument`
+        """
+        return self._repo.set_log_upload_error(_id=doc_id, log_upload_error=log_upload_error)
+
     def delete_operation(self, doc_id):
         """Deletes the given delegated operation.