Support Python 3.14 (#507)

Use uuidv7 by default in Python 3.14.

Also fix an issue where local recovery is attempted while using a
recovery service (Conductor or Cloud), which prints confusing "No
workflows to recover" messages.

Author: kraftp

.github/workflows/unit-test.yml

@@ -22,7 +22,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10', '3.11', '3.12', '3.13']
+        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
         dbos-database: ['SQLITE', 'POSTGRES']
     services:
       # Postgres service container

dbos/_client.py

@@ -1,7 +1,6 @@
 import asyncio
 import json
 import time
-import uuid
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -20,6 +19,7 @@
 from dbos._app_db import ApplicationDatabase
 from dbos._context import MaxPriority, MinPriority
 from dbos._sys_db import SystemDatabase
+from dbos._utils import generate_uuid
 
 if TYPE_CHECKING:
     from dbos._dbos import WorkflowHandle, WorkflowHandleAsync
@@ -188,7 +188,7 @@ def _enqueue(self, options: EnqueueOptions, *args: Any, **kwargs: Any) -> str:
             max_recovery_attempts = DEFAULT_MAX_RECOVERY_ATTEMPTS
         workflow_id = options.get("workflow_id")
         if workflow_id is None:
-            workflow_id = str(uuid.uuid4())
+            workflow_id = generate_uuid()
         workflow_timeout = options.get("workflow_timeout", None)
         enqueue_options_internal: EnqueueOptionsInternal = {
             "deduplication_id": options.get("deduplication_id"),
@@ -281,7 +281,7 @@ def send(
         topic: Optional[str] = None,
         idempotency_key: Optional[str] = None,
     ) -> None:
-        idempotency_key = idempotency_key if idempotency_key else str(uuid.uuid4())
+        idempotency_key = idempotency_key if idempotency_key else generate_uuid()
         status: WorkflowStatusInternal = {
             "workflow_uuid": f"{destination_id}-{idempotency_key}",
             "status": WorkflowStatusString.SUCCESS.value,

dbos/_conductor/conductor.py

@@ -2,7 +2,6 @@
 import threading
 import time
 import traceback
-import uuid
 from importlib.metadata import version
 from typing import TYPE_CHECKING, Optional
 
@@ -11,7 +10,7 @@
 from websockets.sync.connection import Connection
 
 from dbos._context import SetWorkflowID
-from dbos._utils import GlobalParams
+from dbos._utils import GlobalParams, generate_uuid
 from dbos._workflow_commands import (
     garbage_collect,
     get_workflow,
@@ -192,7 +191,7 @@ def run(self) -> None:
                             fork_message = p.ForkWorkflowRequest.from_json(message)
                             new_workflow_id = fork_message.body["new_workflow_id"]
                             if new_workflow_id is None:
-                                new_workflow_id = str(uuid.uuid4())
+                                new_workflow_id = generate_uuid()
                             workflow_id = fork_message.body["workflow_id"]
                             start_step = fork_message.body["start_step"]
                             app_version = fork_message.body["application_version"]

dbos/_context.py

@@ -2,7 +2,6 @@
 
 import json
 import os
-import uuid
 from contextlib import AbstractContextManager
 from contextvars import ContextVar
 from dataclasses import dataclass
@@ -15,7 +14,7 @@
 
 from sqlalchemy.orm import Session
 
-from dbos._utils import GlobalParams
+from dbos._utils import GlobalParams, generate_uuid
 
 from ._logger import dbos_logger
 from ._tracer import dbos_tracer
@@ -151,7 +150,7 @@ def assign_workflow_id(self) -> str:
                 self.logger.warning(
                     f"Multiple workflows started in the same SetWorkflowID block. Only the first workflow is assigned the specified workflow ID; subsequent workflows will use a generated workflow ID."
                 )
-            wfid = str(uuid.uuid4())
+            wfid = generate_uuid()
         return wfid
 
     def start_workflow(

dbos/_dbos.py

@@ -7,7 +7,6 @@
 import sys
 import threading
 import time
-import uuid
 from concurrent.futures import ThreadPoolExecutor
 from logging import Logger
 from typing import (
@@ -33,7 +32,7 @@
 from dbos._debouncer import debouncer_workflow
 from dbos._serialization import DefaultSerializer, Serializer
 from dbos._sys_db import SystemDatabase, WorkflowStatus
-from dbos._utils import INTERNAL_QUEUE_NAME, GlobalParams
+from dbos._utils import INTERNAL_QUEUE_NAME, GlobalParams, generate_uuid
 from dbos._workflow_commands import fork_workflow, list_queued_workflows, list_workflows
 
 from ._classproperty import classproperty
@@ -444,7 +443,7 @@ def _launch(self, *, debug_mode: bool = False) -> None:
             if GlobalParams.app_version == "":
                 GlobalParams.app_version = self._registry.compute_app_version()
             if self.conductor_key is not None:
-                GlobalParams.executor_id = str(uuid.uuid4())
+                GlobalParams.executor_id = generate_uuid()
             dbos_logger.info(f"Executor ID: {GlobalParams.executor_id}")
             dbos_logger.info(f"Application version: {GlobalParams.app_version}")
             self._executor_field = ThreadPoolExecutor(max_workers=sys.maxsize)
@@ -496,20 +495,21 @@ def _launch(self, *, debug_mode: bool = False) -> None:
                 except Exception as e:
                     dbos_logger.warning(f"Failed to start admin server: {e}")
 
-            dbos_logger.debug("Retrieving local pending workflows for recovery")
-            workflow_ids = self._sys_db.get_pending_workflows(
-                GlobalParams.executor_id, GlobalParams.app_version
-            )
-            if (len(workflow_ids)) > 0:
-                self.logger.info(
-                    f"Recovering {len(workflow_ids)} workflows from application version {GlobalParams.app_version}"
-                )
-            else:
-                self.logger.info(
-                    f"No workflows to recover from application version {GlobalParams.app_version}"
+            # Recover local workflows if not using a recovery service
+            if not self.conductor_key and not GlobalParams.dbos_cloud:
+                dbos_logger.debug("Retrieving local pending workflows for recovery")
+                workflow_ids = self._sys_db.get_pending_workflows(
+                    GlobalParams.executor_id, GlobalParams.app_version
                 )
-
-            self._executor.submit(startup_recovery_thread, self, workflow_ids)
+                if (len(workflow_ids)) > 0:
+                    self.logger.info(
+                        f"Recovering {len(workflow_ids)} workflows from application version {GlobalParams.app_version}"
+                    )
+                else:
+                    self.logger.info(
+                        f"No workflows to recover from application version {GlobalParams.app_version}"
+                    )
+                self._executor.submit(startup_recovery_thread, self, workflow_ids)
 
             # Listen to notifications
             dbos_logger.debug("Starting notifications listener thread")

dbos/_dbos_config.py

@@ -265,7 +265,7 @@ def load_config(
             ]
 
     data = cast(ConfigFile, data)
-    return data  # type: ignore
+    return data
 
 
 def process_config(

dbos/_debouncer.py

@@ -2,7 +2,6 @@
 import math
 import time
 import types
-import uuid
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -39,7 +38,7 @@
 from dbos._queue import Queue
 from dbos._registrations import get_dbos_func_name
 from dbos._serialization import WorkflowInputs
-from dbos._utils import INTERNAL_QUEUE_NAME
+from dbos._utils import INTERNAL_QUEUE_NAME, generate_uuid
 
 if TYPE_CHECKING:
     from dbos._dbos import WorkflowHandle, WorkflowHandleAsync
@@ -209,7 +208,7 @@ def debounce(
 
             # Deterministically generate the user workflow ID and message ID
             def assign_debounce_ids() -> tuple[str, str]:
-                return str(uuid.uuid4()), ctx.assign_workflow_id()
+                return generate_uuid(), ctx.assign_workflow_id()
 
             message_id, user_workflow_id = dbos._sys_db.call_function_as_step(
                 assign_debounce_ids, "DBOS.assign_debounce_ids"
@@ -320,14 +319,14 @@ def debounce(
             "workflow_id": (
                 self.workflow_options["workflow_id"]
                 if self.workflow_options.get("workflow_id")
-                else str(uuid.uuid4())
+                else generate_uuid()
             ),
             "app_version": self.workflow_options.get("app_version"),
             "deduplication_id": self.workflow_options.get("deduplication_id"),
             "priority": self.workflow_options.get("priority"),
             "workflow_timeout_sec": self.workflow_options.get("workflow_timeout"),
         }
-        message_id = str(uuid.uuid4())
+        message_id = generate_uuid()
         while True:
             try:
                 # Attempt to enqueue a debouncer for this workflow.

dbos/_fastapi.py

@@ -1,4 +1,3 @@
-import uuid
 from typing import Any, Callable, MutableMapping, cast
 
 from fastapi import FastAPI
@@ -9,15 +8,15 @@
 from . import DBOS
 from ._context import EnterDBOSHandler, OperationType, SetWorkflowID, TracedAttributes
 from ._error import DBOSException
-from ._utils import request_id_header
+from ._utils import generate_uuid, request_id_header
 
 
 def _get_or_generate_request_id(request: FastAPIRequest) -> str:
     request_id = request.headers.get(request_id_header, None)
     if request_id is not None:
         return request_id
     else:
-        return str(uuid.uuid4())
+        return generate_uuid()
 
 
 async def _dbos_error_handler(request: FastAPIRequest, gexc: Exception) -> JSONResponse:

dbos/_flask.py

@@ -1,12 +1,11 @@
-import uuid
 from typing import Any
 from urllib.parse import urlparse
 
 from flask import Flask
 from werkzeug.wrappers import Request as WRequest
 
 from ._context import EnterDBOSHandler, OperationType, SetWorkflowID, TracedAttributes
-from ._utils import request_id_header
+from ._utils import generate_uuid, request_id_header
 
 
 class FlaskMiddleware:
@@ -41,7 +40,7 @@ def _get_or_generate_request_id(request: WRequest) -> str:
     if request_id is not None:
         return request_id
     else:
-        return str(uuid.uuid4())
+        return generate_uuid()
 
 
 def setup_flask_middleware(app: Flask) -> None:

dbos/_utils.py

@@ -1,5 +1,7 @@
 import importlib.metadata
 import os
+import sys
+import uuid
 
 import psycopg
 from sqlalchemy.exc import DBAPIError
@@ -12,6 +14,7 @@
 class GlobalParams:
     app_version: str = os.environ.get("DBOS__APPVERSION", "")
     executor_id: str = os.environ.get("DBOS__VMID", "local")
+    dbos_cloud: bool = os.environ.get("DBOS__CLOUD") == "true"
     try:
         # Only works on Python >= 3.8
         dbos_version = importlib.metadata.version("dbos")
@@ -57,3 +60,10 @@ def retriable_sqlite_exception(e: Exception) -> bool:
         return True
     else:
         return False
+
+
+def generate_uuid() -> str:
+    if sys.version_info >= (3, 14):
+        return str(uuid.uuid7())
+    else:
+        return str(uuid.uuid4())