Scheduler/default scheduled

fiftyone/core/cli.py

@@ -7,7 +7,6 @@
 """
 
 import argparse
-import warnings
 from collections import defaultdict
 from datetime import datetime
 import json
@@ -19,8 +18,6 @@
 
 import argcomplete
 from bson import ObjectId
-import humanize
-import pytz
 from tabulate import tabulate
 import webbrowser
 

fiftyone/factory/repos/delegated_operation.py

@@ -15,6 +15,7 @@
 from pymongo import IndexModel
 from pymongo.collection import Collection
 
+from fiftyone.internal.util import is_remote_service
 from fiftyone.factory import DelegatedOperationPagingParams
 from fiftyone.factory.repos import DelegatedOperationDocument
 from fiftyone.operators.executor import (
@@ -133,7 +134,7 @@ def __init__(self, collection: Collection = None):
         self._collection = (
             collection if collection is not None else self._get_collection()
         )
-
+        self.is_remote = is_remote_service()
         self._create_indexes()
 
     def _get_collection(self) -> Collection:
@@ -169,7 +170,7 @@ def _create_indexes(self):
             self._collection.create_indexes(indices_to_create)
 
     def queue_operation(self, **kwargs: Any) -> DelegatedOperationDocument:
-        op = DelegatedOperationDocument()
+        op = DelegatedOperationDocument(is_remote=self.is_remote)
         for prop in self.required_props:
             if prop not in kwargs:
                 raise ValueError("Missing required property '%s'" % prop)

fiftyone/factory/repos/delegated_operation_doc.py

@@ -25,6 +25,7 @@ def __init__(
         operator: str = None,
         delegation_target: str = None,
         context: dict = None,
+        is_remote: bool = False,
     ):
         self.operator = operator
         self.label = None
@@ -35,18 +36,20 @@ def __init__(
             else context
         )
         self.run_state = (
-            ExecutionRunState.QUEUED
-        )  # default to queued state on create
+            ExecutionRunState.SCHEDULED
+            if is_remote
+            else ExecutionRunState.QUEUED
+        )  # if running locally use QUEUED otherwise SCHEDULED
         self.run_link = None
-        self.queued_at = datetime.utcnow()
+        self.queued_at = datetime.utcnow() if not is_remote else None
         self.updated_at = datetime.utcnow()
         self.status = None
         self.dataset_id = None
         self.started_at = None
         self.pinned = False
         self.completed_at = None
         self.failed_at = None
-        self.scheduled_at = None
+        self.scheduled_at = datetime.utcnow() if is_remote else None
         self.result = None
         self.id = None
         self._doc = None

fiftyone/internal/__init__.py

@@ -7,3 +7,4 @@
 """
 
 from .secrets import *
+from .util import is_remote_service

fiftyone/internal/util.py

@@ -0,0 +1,34 @@
+"""
+FiftyOne internal utilities.
+
+| Copyright 2017-2024, Voxel51, Inc.
+| `voxel51.com <https://voxel51.com/>`_
+|
+"""
+
+
+def is_remote_service():
+    """Whether the SDK is running in a remote service context.
+
+    Returns:
+        True/False
+    """
+    return has_encryption_key() and has_api_key()
+
+
+def has_encryption_key():
+    """Whether the current environment has an encryption key.
+
+    Returns:
+        True/False
+    """
+    return False
+
+
+def has_api_key():
+    """Whether the current environment has an API key.
+
+    Returns:
+        True/False
+    """
+    return False

tests/unittests/delegated_operators_tests.py

@@ -309,9 +309,6 @@ def test_list_operations(self, mock_get_operator, mock_operator_exists):
         for doc in docs_to_run:
             self.svc.set_scheduled(doc)
 
-        queued = self.svc.get_queued_operations()
-        self.assertEqual(len(queued), 10 + initial_queued)
-
         scheduled = self.svc.get_scheduled_operations()
         self.assertEqual(len(scheduled), 10 + initial_scheduled)