add scheduled state for delegated_ops collection

docs/source/cli/index.rst

@@ -989,9 +989,9 @@ List delegated operations.
                             only list operations for this dataset
       -s STATE, --state STATE
                             only list operations with this state. Supported
-                            values are ('QUEUED', 'RUNNING', 'COMPLETED', 'FAILED')
+                            values are ('QUEUED', 'PENDING', 'RUNNING', 'COMPLETED', 'FAILED')
       --sort-by SORT_BY     how to sort the operations. Supported values are
-                            ('QUEUED_AT', 'STARTED_AT', COMPLETED_AT', 'FAILED_AT', 'OPERATOR')
+                            ('QUEUED_AT', 'PENDING_AT', 'STARTED_AT', COMPLETED_AT', 'FAILED_AT', 'OPERATOR')
       --reverse             whether to sort in reverse order
       -l LIMIT, --limit LIMIT
                             a maximum number of operations to show

fiftyone/core/cli.py

@@ -3150,15 +3150,15 @@ def setup(parser):
             default=None,
             help=(
                 "only list operations with this state. Supported values are "
-                "('QUEUED', 'RUNNING', 'COMPLETED', 'FAILED')"
+                "('QUEUED', 'PENDING', 'RUNNING', 'COMPLETED', 'FAILED')"
             ),
         )
         parser.add_argument(
             "--sort-by",
             default="QUEUED_AT",
             help=(
                 "how to sort the operations. Supported values are "
-                "('QUEUED_AT', 'STARTED_AT', COMPLETED_AT', 'FAILED_AT', 'OPERATOR')"
+                "('QUEUED_AT', 'PENDING_AT', 'STARTED_AT', COMPLETED_AT', 'FAILED_AT', 'OPERATOR')"
             ),
         )
         parser.add_argument(
@@ -3390,7 +3390,7 @@ def setup(parser):
             default=None,
             help=(
                 "delete operations in this state. Supported values are "
-                "('QUEUED', 'COMPLETED', 'FAILED')"
+                "('QUEUED', 'PENDING', 'COMPLETED', 'FAILED')"
             ),
         )
         parser.add_argument(

fiftyone/factory/__init__.py

@@ -12,6 +12,7 @@ class SortByField(object):
 
     UPDATED_AT = "updated_at"
     QUEUED_AT = "queued_at"
+    PENDING_AT = "pending_at"
     COMPLETED_AT = "completed_at"
     STARTED_AT = "started_at"
     FAILED_AT = "failed_at"

fiftyone/factory/repos/delegated_operation.py

@@ -64,6 +64,22 @@ def get_queued_operations(
             "subclass must implement get_queued_operations()"
         )
 
+    def get_pending_operations(
+        self, operator: str = None, dataset_name=None
+    ) -> List[DelegatedOperationDocument]:
+        """Get all pending operations."""
+        raise NotImplementedError(
+            "subclass must implement get_pending_operations()"
+        )
+
+    def get_running_operations(
+        self, operator: str = None, dataset_name=None
+    ) -> List[DelegatedOperationDocument]:
+        """Get all running operations."""
+        raise NotImplementedError(
+            "subclass must implement get_running_operations()"
+        )
+
     def list_operations(
         self,
         operator: str = None,
@@ -275,6 +291,14 @@ def update_run_state(
                     "updated_at": datetime.utcnow(),
                 }
             }
+        elif run_state == ExecutionRunState.PENDING:
+            update = {
+                "$set": {
+                    "run_state": run_state,
+                    "pending_at": datetime.utcnow(),
+                    "updated_at": datetime.utcnow(),
+                }
+            }
 
         if run_link is not None:
             update["$set"]["run_link"] = run_link
@@ -341,6 +365,28 @@ def get_queued_operations(
             run_state=ExecutionRunState.QUEUED,
         )
 
+    def get_pending_operations(
+        self,
+        operator: str = None,
+        dataset_name: ObjectId = None,
+    ) -> List[DelegatedOperationDocument]:
+        return self.list_operations(
+            operator=operator,
+            dataset_name=dataset_name,
+            run_state=ExecutionRunState.PENDING,
+        )
+
+    def get_running_operations(
+        self,
+        operator: str = None,
+        dataset_name: ObjectId = None,
+    ) -> List[DelegatedOperationDocument]:
+        return self.list_operations(
+            operator=operator,
+            dataset_name=dataset_name,
+            run_state=ExecutionRunState.RUNNING,
+        )
+
     def list_operations(
         self,
         operator: str = None,

fiftyone/factory/repos/delegated_operation_doc.py

@@ -46,6 +46,7 @@ def __init__(
         self.pinned = False
         self.completed_at = None
         self.failed_at = None
+        self.pending_at = None
         self.result = None
         self.id = None
         self._doc = None
@@ -68,6 +69,7 @@ def from_pymongo(self, doc: dict):
             doc["completed_at"] if "completed_at" in doc else None
         )
         self.failed_at = doc["failed_at"] if "failed_at" in doc else None
+        self.pending_at = doc["pending_at"] if "pending_at" in doc else None
         self.pinned = doc["pinned"] if "pinned" in doc else None
         self.dataset_id = doc["dataset_id"] if "dataset_id" in doc else None
         self.run_link = doc["run_link"] if "run_link" in doc else None

fiftyone/operators/delegated.py

@@ -101,6 +101,17 @@ def set_running(self, doc_id, progress=None, run_link=None):
             progress=progress,
         )
 
+    def set_pending(self, doc_id):
+        """Sets the given delegated operation to pending state.
+        Args:
+            doc_id: the ID of the delegated operation
+        Returns:
+            a :class:`fiftyone.factory.repos.DelegatedOperationDocument`
+        """
+        return self._repo.update_run_state(
+            _id=doc_id, run_state=ExecutionRunState.PENDING
+        )
+
     def set_completed(
         self,
         doc_id,
@@ -235,6 +246,34 @@ def get_queued_operations(self, operator=None, dataset_name=None):
             operator=operator, dataset_name=dataset_name
         )
 
+    def get_pending_operations(self, operator=None, dataset_name=None):
+        """Returns all pending delegated operations.
+        Args:
+            operator (None): the optional name of the operator to return all
+                the pending delegated operations for
+            dataset_name (None): the optional name of the dataset to return all
+                the pending delegated operations for
+        Returns:
+            a list of :class:`fiftyone.factory.repos.DelegatedOperationDocument`
+        """
+        return self._repo.get_pending_operations(
+            operator=operator, dataset_name=dataset_name
+        )
+
+    def get_running_operations(self, operator=None, dataset_name=None):
+        """Returns all running delegated operations.
+        Args:
+            operator (None): the optional name of the operator to return all
+                the running delegated operations for
+            dataset_name (None): the optional name of the dataset to return all
+                the running delegated operations for
+        Returns:
+            a list of :class:`fiftyone.factory.repos.DelegatedOperationDocument`
+        """
+        return self._repo.get_running_operations(
+            operator=operator, dataset_name=dataset_name
+        )
+
     def get(self, doc_id):
         """Returns the delegated operation with the given ID.
 

fiftyone/operators/executor.py

@@ -37,6 +37,7 @@ class ExecutionRunState(object):
     """Enumeration of the available operator run states."""
 
     QUEUED = "queued"
+    PENDING = "pending"
     RUNNING = "running"
     COMPLETED = "completed"
     FAILED = "failed"