Fixes and improvements

- Add new_uuid method to OrchestrationContext for deterministic replay-safe UUIDs
- Fix entity locking behavior for Functions
- Align _RuntimeOrchestrationContext param names with OrchestrationContext
- Remap __init__.py files for new module
- Update version to 0.0.1dev0
- Add docstrings to missing methods
- Move code for executing orchestrators/entities to DurableFunctionsWorker
- Add function metadata to triggers for detection by extension

Author: andystaples

durabletask-azurefunctions/durabletask/azurefunctions/__init__.py

@@ -1,2 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
+
+from durabletask.azurefunctions.decorators.durable_app import Blueprint, DFApp
+from durabletask.azurefunctions.client import DurableFunctionsClient
+
+__all__ = ["Blueprint", "DFApp", "DurableFunctionsClient"]

durabletask-azurefunctions/durabletask/azurefunctions/client.py

@@ -16,6 +16,12 @@
 
 # Client class used for Durable Functions
 class DurableFunctionsClient(TaskHubGrpcClient):
+    """A gRPC client passed to Durable Functions durable client bindings.
+
+    Connects to the Durable Functions runtime using gRPC and provides methods
+    for creating and managing Durable orchestrations, interacting with Durable entities, 
+    and creating HTTP management payloads and check status responses for use with Durable Functions invocations.
+    """
     taskHubName: str
     connectionName: str
     creationUrls: dict[str, str]
@@ -28,6 +34,16 @@ class DurableFunctionsClient(TaskHubGrpcClient):
     grpcHttpClientTimeout: timedelta
 
     def __init__(self, client_as_string: str):
+        """Initializes a DurableFunctionsClient instance from a JSON string.
+
+        This string will be provided by the Durable Functions host extension upon invocation of the client trigger.
+
+        Args:
+            client_as_string (str): A JSON string containing the Durable Functions client configuration.
+
+        Raises:
+            json.JSONDecodeError: If the provided string is not valid JSON.
+        """
         client = json.loads(client_as_string)
 
         self.taskHubName = client.get("taskHubName", "")

durabletask-azurefunctions/durabletask/azurefunctions/constants.py

@@ -2,11 +2,6 @@
 # Licensed under the MIT License.
 
 """Constants used to determine the local running context."""
-# TODO: Remove unused constants after module is complete
-DEFAULT_LOCAL_HOST: str = 'localhost:7071'
-DEFAULT_LOCAL_ORIGIN: str = f'http://{DEFAULT_LOCAL_HOST}'
-DATETIME_STRING_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
-HTTP_ACTION_NAME = 'BuiltIn::HttpActivity'
 ORCHESTRATION_TRIGGER = "orchestrationTrigger"
 ACTIVITY_TRIGGER = "activityTrigger"
 ENTITY_TRIGGER = "entityTrigger"

durabletask-azurefunctions/durabletask/azurefunctions/decorators/__init__.py

@@ -1,11 +1,2 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
-
-"""Durable Task SDK for Python entities component"""
-
-import durabletask.azurefunctions.decorators.durable_app as durable_app
-import durabletask.azurefunctions.decorators.metadata as metadata
-
-__all__ = ["durable_app", "metadata"]
-
-PACKAGE_NAME = "durabletask.azurefunctions.decorators"

durabletask-azurefunctions/durabletask/azurefunctions/decorators/durable_app.py

@@ -1,20 +1,16 @@
 #  Copyright (c) Microsoft Corporation. All rights reserved.
 #  Licensed under the MIT License.
 
-import base64
 from functools import wraps
 
-from durabletask.internal.orchestrator_service_pb2 import EntityRequest, EntityBatchRequest, EntityBatchResult, OrchestratorRequest, OrchestratorResponse
 from .metadata import OrchestrationTrigger, ActivityTrigger, EntityTrigger, \
     DurableClient
 from typing import Callable, Optional
 from typing import Union
 from azure.functions import FunctionRegister, TriggerApi, BindingApi, AuthLevel
 
-# TODO: Use __init__.py to optimize imports
 from durabletask.azurefunctions.client import DurableFunctionsClient
 from durabletask.azurefunctions.worker import DurableFunctionsWorker
-from durabletask.azurefunctions.internal.azurefunctions_null_stub import AzureFunctionsNullStub
 
 
 class Blueprint(TriggerApi, BindingApi):
@@ -61,40 +57,8 @@ def _configure_orchestrator_callable(self, wrap) -> Callable:
         def decorator(orchestrator_func):
             # Construct an orchestrator based on the end-user code
 
-            # TODO: Move this logic somewhere better
             def handle(context) -> str:
-                context_body = getattr(context, "body", None)
-                if context_body is None:
-                    context_body = context
-                orchestration_context = context_body
-                request = OrchestratorRequest()
-                request.ParseFromString(base64.b64decode(orchestration_context))
-                stub = AzureFunctionsNullStub()
-                worker = DurableFunctionsWorker()
-                response: Optional[OrchestratorResponse] = None
-
-                def stub_complete(stub_response):
-                    nonlocal response
-                    response = stub_response
-                stub.CompleteOrchestratorTask = stub_complete
-                execution_started_events = []
-                for e in request.pastEvents:
-                    if e.HasField("executionStarted"):
-                        execution_started_events.append(e)
-                for e in request.newEvents:
-                    if e.HasField("executionStarted"):
-                        execution_started_events.append(e)
-                if len(execution_started_events) == 0:
-                    raise Exception("No ExecutionStarted event found in orchestration request.")
-
-                function_name = execution_started_events[-1].executionStarted.name
-                worker.add_named_orchestrator(function_name, orchestrator_func)
-                worker._execute_orchestrator(request, stub, None)
-
-                if response is None:
-                    raise Exception("Orchestrator execution did not produce a response.")
-                # The Python worker returns the input as type "json", so double-encoding is necessary
-                return '"' + base64.b64encode(response.SerializeToString()).decode('utf-8') + '"'
+                return DurableFunctionsWorker()._execute_orchestrator(orchestrator_func, context)
 
             handle.orchestrator_function = orchestrator_func  # type: ignore
 
@@ -121,33 +85,11 @@ def _configure_entity_callable(self, wrap) -> Callable:
         def decorator(entity_func):
             # Construct an orchestrator based on the end-user code
 
-            # TODO: Move this logic somewhere better
             # TODO: Because this handle method is the one actually exposed to the Functions SDK decorator,
             #       the parameter name will always be "context" here, even if the user specified a different name.
             #       We need to find a way to allow custom context names (like "ctx").
             def handle(context) -> str:
-                context_body = getattr(context, "body", None)
-                if context_body is None:
-                    context_body = context
-                orchestration_context = context_body
-                request = EntityBatchRequest()
-                request.ParseFromString(base64.b64decode(orchestration_context))
-                stub = AzureFunctionsNullStub()
-                worker = DurableFunctionsWorker()
-                response: Optional[EntityBatchResult] = None
-
-                def stub_complete(stub_response: EntityBatchResult):
-                    nonlocal response
-                    response = stub_response
-                stub.CompleteEntityTask = stub_complete
-
-                worker.add_entity(entity_func)
-                worker._execute_entity_batch(request, stub, None)
-
-                if response is None:
-                    raise Exception("Entity execution did not produce a response.")
-                # The Python worker returns the input as type "json", so double-encoding is necessary
-                return '"' + base64.b64encode(response.SerializeToString()).decode('utf-8') + '"'
+                return DurableFunctionsWorker()._execute_entity_batch(entity_func, context)
 
             handle.entity_function = entity_func  # type: ignore
 
@@ -157,8 +99,7 @@ def stub_complete(stub_response: EntityBatchResult):
 
         return decorator
 
-    def _add_rich_client(self, fb, parameter_name,
-                         client_constructor):
+    def _add_rich_client(self, fb, parameter_name, client_constructor):
         # Obtain user-code and force type annotation on the client-binding parameter to be `str`.
         # This ensures a passing type-check of that specific parameter,
         # circumventing a limitation of the worker in type-checking rich DF Client objects.

durabletask-azurefunctions/durabletask/azurefunctions/decorators/metadata.py

@@ -28,6 +28,7 @@ def get_binding_name() -> str:
     def __init__(self,
                  name: str,
                  orchestration: Optional[str] = None,
+                 durable_requires_grpc=True,
                  ) -> None:
         self.orchestration = orchestration
         super().__init__(name=name)
@@ -53,6 +54,7 @@ def get_binding_name() -> str:
     def __init__(self,
                  name: str,
                  activity: Optional[str] = None,
+                 durable_requires_grpc=True,
                  ) -> None:
         self.activity = activity
         super().__init__(name=name)
@@ -78,6 +80,7 @@ def get_binding_name() -> str:
     def __init__(self,
                  name: str,
                  entity_name: Optional[str] = None,
+                 durable_requires_grpc=True,
                  ) -> None:
         self.entity_name = entity_name
         super().__init__(name=name)
@@ -103,7 +106,8 @@ def get_binding_name() -> str:
     def __init__(self,
                  name: str,
                  task_hub: Optional[str] = None,
-                 connection_name: Optional[str] = None
+                 connection_name: Optional[str] = None,
+                 durable_requires_grpc=True,
                  ) -> None:
         self.task_hub = task_hub
         self.connection_name = connection_name

durabletask-azurefunctions/durabletask/azurefunctions/http/http_management_payload.py

@@ -2,7 +2,20 @@
 
 
 class HttpManagementPayload:
+    """A class representing the HTTP management payload for a Durable Function orchestration instance.
+
+    Contains URLs for managing the instance, such as querying status,
+    sending events, terminating, restarting, etc.
+    """
+
     def __init__(self, instance_id: str, instance_status_url: str, required_query_string_parameters: str):
+        """Initializes the HttpManagementPayload with the necessary URLs.
+
+        Args:
+            instance_id (str): The ID of the Durable Function instance.
+            instance_status_url (str): The base URL for the instance status.
+            required_query_string_parameters (str): The required URL parameters provided by the Durable extension.
+        """
         self.urls = {
             'id': instance_id,
             'purgeHistoryDeleteUri': instance_status_url + "?" + required_query_string_parameters,

durabletask-azurefunctions/durabletask/azurefunctions/worker.py

@@ -1,15 +1,22 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import base64
 from threading import Event
+from typing import Optional
+from durabletask.internal.orchestrator_service_pb2 import EntityBatchRequest, EntityBatchResult, OrchestratorRequest, OrchestratorResponse
 from durabletask.worker import _Registry, ConcurrencyOptions
 from durabletask.internal import shared
 from durabletask.worker import TaskHubGrpcWorker
+from durabletask.azurefunctions.internal.azurefunctions_null_stub import AzureFunctionsNullStub
 
 
 # Worker class used for Durable Task Scheduler (DTS)
 class DurableFunctionsWorker(TaskHubGrpcWorker):
-    """TODO: Docs
+    """A worker that can execute orchestrator and entity functions in the context of Azure Functions.
+
+    Used internally by the Durable Functions Python SDK, and should not be visible to functionapps directly.
+    See TaskHubGrpcWorker for base class documentation.
     """
 
     def __init__(self):
@@ -27,6 +34,60 @@ def __init__(self):
         self._interceptors = None
 
     def add_named_orchestrator(self, name: str, func):
-        """TODO: Docs
-        """
         self._registry.add_named_orchestrator(name, func)
+
+    def _execute_orchestrator(self, func, context) -> str:
+        context_body = getattr(context, "body", None)
+        if context_body is None:
+            context_body = context
+        orchestration_context = context_body
+        request = OrchestratorRequest()
+        request.ParseFromString(base64.b64decode(orchestration_context))
+        stub = AzureFunctionsNullStub()
+        response: Optional[OrchestratorResponse] = None
+
+        def stub_complete(stub_response):
+            nonlocal response
+            response = stub_response
+        stub.CompleteOrchestratorTask = stub_complete
+        execution_started_events = []
+        for e in request.pastEvents:
+            if e.HasField("executionStarted"):
+                execution_started_events.append(e)
+        for e in request.newEvents:
+            if e.HasField("executionStarted"):
+                execution_started_events.append(e)
+        if len(execution_started_events) == 0:
+            raise Exception("No ExecutionStarted event found in orchestration request.")
+
+        function_name = execution_started_events[-1].executionStarted.name
+        self.add_named_orchestrator(function_name, func)
+        super()._execute_orchestrator(request, stub, None)
+
+        if response is None:
+            raise Exception("Orchestrator execution did not produce a response.")
+        # The Python worker returns the input as type "json", so double-encoding is necessary
+        return '"' + base64.b64encode(response.SerializeToString()).decode('utf-8') + '"'
+
+    def _execute_entity_batch(self, func, context) -> str:
+        context_body = getattr(context, "body", None)
+        if context_body is None:
+            context_body = context
+        orchestration_context = context_body
+        request = EntityBatchRequest()
+        request.ParseFromString(base64.b64decode(orchestration_context))
+        stub = AzureFunctionsNullStub()
+        response: Optional[EntityBatchResult] = None
+
+        def stub_complete(stub_response: EntityBatchResult):
+            nonlocal response
+            response = stub_response
+        stub.CompleteEntityTask = stub_complete
+
+        self.add_entity(func)
+        super()._execute_entity_batch(request, stub, None)
+
+        if response is None:
+            raise Exception("Entity execution did not produce a response.")
+        # The Python worker returns the input as type "json", so double-encoding is necessary
+        return '"' + base64.b64encode(response.SerializeToString()).decode('utf-8') + '"'

durabletask-azurefunctions/pyproject.toml

@@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durabletask.azurefunctions"
-version = "0.1.0"
+version = "0.0.1dev0"
 description = "Durable Task Python SDK provider implementation for Durable Azure Functions"
 keywords = [
     "durable",

durabletask/internal/helpers.py

@@ -4,7 +4,6 @@
 import traceback
 from datetime import datetime
 from typing import Optional
-import uuid
 
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
@@ -197,8 +196,11 @@ def new_schedule_task_action(id: int, name: str, encoded_input: Optional[str],
     ))
 
 
-def new_call_entity_action(id: int, parent_instance_id: str, entity_id: EntityInstanceId, operation: str, encoded_input: Optional[str]):
-    request_id = str(uuid.uuid4())
+def new_call_entity_action(id: int,
+                           parent_instance_id: str,
+                           entity_id: EntityInstanceId,
+                           operation: str, encoded_input: Optional[str],
+                           request_id: str):
     return pb.OrchestratorAction(id=id, sendEntityMessage=pb.SendEntityMessageAction(entityOperationCalled=pb.EntityOperationCalledEvent(
         requestId=request_id,
         operation=operation,
@@ -210,8 +212,11 @@ def new_call_entity_action(id: int, parent_instance_id: str, entity_id: EntityIn
     )))
 
 
-def new_signal_entity_action(id: int, entity_id: EntityInstanceId, operation: str, encoded_input: Optional[str]):
-    request_id = str(uuid.uuid4())
+def new_signal_entity_action(id: int,
+                             entity_id: EntityInstanceId,
+                             operation: str,
+                             encoded_input: Optional[str],
+                             request_id: str):
     return pb.OrchestratorAction(id=id, sendEntityMessage=pb.SendEntityMessageAction(entityOperationSignaled=pb.EntityOperationSignaledEvent(
         requestId=request_id,
         operation=operation,