Application Version and Executor ID in DBOSConfig (#417)

Addresses #411

Author: kraftp

dbos/_dbos.py

@@ -324,6 +324,17 @@ def __init__(
         self.conductor_websocket: Optional[ConductorWebsocket] = None
         self._background_event_loop: BackgroundEventLoop = BackgroundEventLoop()
 
+        # Globally set the application version and executor ID.
+        # In DBOS Cloud, instead use the values supplied through environment variables.
+        if not os.environ.get("DBOS__CLOUD") == "true":
+            if (
+                "application_version" in config
+                and config["application_version"] is not None
+            ):
+                GlobalParams.app_version = config["application_version"]
+            if "executor_id" in config and config["executor_id"] is not None:
+                GlobalParams.executor_id = config["executor_id"]
+
         init_logger()
 
         # Translate user provided config to an internal format

dbos/_dbos_config.py

@@ -47,6 +47,8 @@ class DBOSConfig(TypedDict, total=False):
     admin_port: Optional[int]
     run_admin_server: Optional[bool]
     otlp_attributes: Optional[dict[str, str]]
+    application_version: Optional[str]
+    executor_id: Optional[str]
 
 
 class RuntimeConfig(TypedDict, total=False):

tests/test_dbos.py

@@ -1397,7 +1397,7 @@ def workflow_one(x: int) -> int:
     assert GlobalParams.app_version != app_version
 
     # Verify that version can be overriden with an environment variable
-    app_version = "12345"
+    app_version = str(uuid.uuid4())
     os.environ["DBOS__APPVERSION"] = app_version
 
     DBOS.destroy(destroy_registry=True)
@@ -1412,6 +1412,28 @@ def workflow_one(x: int) -> int:
 
     del os.environ["DBOS__APPVERSION"]
 
+    # Verify that version and executor ID can be overriden with a config parameter
+    app_version = str(uuid.uuid4())
+    executor_id = str(uuid.uuid4())
+
+    DBOS.destroy(destroy_registry=True)
+    config["application_version"] = app_version
+    config["executor_id"] = executor_id
+    DBOS(config=config)
+
+    @DBOS.workflow()
+    def test_workflow() -> str:
+        assert DBOS.workflow_id
+        return DBOS.workflow_id
+
+    DBOS.launch()
+    assert GlobalParams.app_version == app_version
+    assert GlobalParams.executor_id == executor_id
+    wfid = test_workflow()
+    handle: WorkflowHandle[str] = DBOS.retrieve_workflow(wfid)
+    assert handle.get_status().app_version == app_version
+    assert handle.get_status().executor_id == executor_id
+
 
 def test_recovery_appversion(config: DBOSConfig) -> None:
     input = 5