Working orchestrators + activities

andystaples · andystaples · commit 552a2ddbe1d0 · 2025-11-21T13:46:16.000-07:00
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/client.py b/durabletask-azurefunctions/durabletask/azurefunctions/client.py
@@ -0,0 +1,85 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import json
+
+from datetime import timedelta
+from typing import Any, Optional
+import azure.functions as func
+
+from durabletask.entities import EntityInstanceId
+from durabletask.client import TaskHubGrpcClient
+from durabletask.azurefunctions.internal.azurefunctions_grpc_interceptor import AzureFunctionsDefaultClientInterceptorImpl
+
+
+# Client class used for Durable Functions
+class DurableFunctionsClient(TaskHubGrpcClient):
+    taskHubName: str
+    connectionName: str
+    creationUrls: dict[str, str]
+    managementUrls: dict[str, str]
+    baseUrl: str
+    requiredQueryStringParameters: str
+    rpcBaseUrl: str
+    httpBaseUrl: str
+    maxGrpcMessageSizeInBytes: int
+    grpcHttpClientTimeout: timedelta
+
+    def __init__(self, client_as_string: str):
+        client = json.loads(client_as_string)
+
+        self.taskHubName = client.get("taskHubName", "")
+        self.connectionName = client.get("connectionName", "")
+        self.creationUrls = client.get("creationUrls", {})
+        self.managementUrls = client.get("managementUrls", {})
+        self.baseUrl = client.get("baseUrl", "")
+        self.requiredQueryStringParameters = client.get("requiredQueryStringParameters", "")
+        self.rpcBaseUrl = client.get("rpcBaseUrl", "")
+        self.httpBaseUrl = client.get("httpBaseUrl", "")
+        self.maxGrpcMessageSizeInBytes = client.get("maxGrpcMessageSizeInBytes", 0)
+        # TODO: convert the string value back to timedelta - annoying regex?
+        self.grpcHttpClientTimeout = client.get("grpcHttpClientTimeout", timedelta(seconds=30))
+        interceptors = [AzureFunctionsDefaultClientInterceptorImpl(self.taskHubName, self.requiredQueryStringParameters)]
+
+        # We pass in None for the metadata so we don't construct an additional interceptor in the parent class
+        # Since the parent class doesn't use anything metadata for anything else, we can set it as None
+        super().__init__(
+            host_address=self.rpcBaseUrl,
+            secure_channel=False,
+            metadata=None,
+            interceptors=interceptors)
+
+    def create_check_status_response(self, request: func.HttpRequest, instance_id: str) -> func.HttpResponse:
+        """Creates an HTTP response for checking the status of a Durable Function instance.
+
+        Args:
+            request (func.HttpRequest): The incoming HTTP request.
+            instance_id (str): The ID of the Durable Function instance.
+        """
+        raise NotImplementedError("This method is not implemented yet.")
+
+    def create_http_management_payload(self, instance_id: str) -> dict[str, str]:
+        """Creates an HTTP management payload for a Durable Function instance.
+
+        Args:
+            instance_id (str): The ID of the Durable Function instance.
+        """
+        raise NotImplementedError("This method is not implemented yet.")
+
+    def read_entity_state(
+        self,
+        entity_id: EntityInstanceId,
+        task_hub_name: Optional[str],
+        connection_name: Optional[str]
+    ) -> tuple[bool, Any]:
+        """Reads the state of a Durable Entity.
+
+        Args:
+            entity_id (str): The ID of the Durable Entity.
+            task_hub_name (Optional[str]): The name of the task hub.
+            connection_name (Optional[str]): The name of the connection.
+
+        Returns:
+            (bool, Any): A tuple containing a boolean indicating if the entity exists and its state.
+        """
+        raise NotImplementedError("This method is not implemented yet.")
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/constants.py b/durabletask-azurefunctions/durabletask/azurefunctions/constants.py
@@ -1,5 +1,5 @@
 """Constants used to determine the local running context."""
-# Todo: Remove unused constants after module is complete
+# 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'
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/decorators/durable_app.py b/durabletask-azurefunctions/durabletask/azurefunctions/decorators/durable_app.py
@@ -1,10 +1,19 @@
 #  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 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, OrchestrationContext
+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):
@@ -37,9 +46,6 @@ def __init__(self,
     def _configure_orchestrator_callable(self, wrap) -> Callable:
         """Obtain decorator to construct an Orchestrator class from a user-defined Function.
 
-        In the old programming model, this decorator's logic was unavoidable boilerplate
-        in user-code. Now, this is handled internally by the framework.
-
         Parameters
         ----------
         wrap: Callable
@@ -54,14 +60,31 @@ def _configure_orchestrator_callable(self, wrap) -> Callable:
         def decorator(orchestrator_func):
             # Construct an orchestrator based on the end-user code
 
-            # TODO: Extract this logic (?)
-            def handle(context: OrchestrationContext) -> str:
+            # 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
-                # TODO: Run the orchestration using the context
-                return ""
+                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 = [e for e in [e1 for e1 in request.newEvents] + [e2 for e2 in request.pastEvents] if e.HasField("executionStarted")]
+                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') + '"'
 
             handle.orchestrator_function = orchestrator_func
 
@@ -71,6 +94,55 @@ def handle(context: OrchestrationContext) -> str:
 
         return decorator
 
+    def _configure_entity_callable(self, wrap) -> Callable:
+        """Obtain decorator to construct an Entity class from a user-defined Function.
+
+        Parameters
+        ----------
+        wrap: Callable
+            The next decorator to be applied.
+
+        Returns
+        -------
+        Callable
+            The function to construct an Entity class from the user-defined Function,
+            wrapped by the next decorator in the sequence.
+        """
+        def decorator(entity_func):
+            # TODO: Implement entity support - similar to orchestrators (?)
+            raise NotImplementedError()
+
+        return decorator
+
+    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.
+        # TODO: Once rich-binding type checking is possible, remove the annotation change.
+        user_code = fb._function._func
+        user_code.__annotations__[parameter_name] = str
+
+        # `wraps` This ensures we re-export the same method-signature as the decorated method
+        @wraps(user_code)
+        async def df_client_middleware(*args, **kwargs):
+
+            # Obtain JSON-string currently passed as DF Client,
+            # construct rich object from it,
+            # and assign parameter to that rich object
+            starter = kwargs[parameter_name]
+            client = client_constructor(starter)
+            kwargs[parameter_name] = client
+
+            # Invoke user code with rich DF Client binding
+            return await user_code(*args, **kwargs)
+
+        # TODO: Is there a better way to support retrieving the unwrapped user code?
+        df_client_middleware.client_function = fb._function._func  # type: ignore
+
+        user_code_with_rich_client = df_client_middleware
+        fb._function._func = user_code_with_rich_client
+
     def orchestration_trigger(self, context_name: str,
                               orchestration: Optional[str] = None):
         """Register an Orchestrator Function.
@@ -133,6 +205,7 @@ def entity_trigger(self, context_name: str,
             Name of Entity Function.
             The value is None by default, in which case the name of the method is used.
         """
+        @self._configure_entity_callable
         @self._configure_function_builder
         def wrap(fb):
             def decorator():
@@ -171,7 +244,7 @@ def durable_client_input(self,
         @self._configure_function_builder
         def wrap(fb):
             def decorator():
-                # self._add_rich_client(fb, client_name, DurableOrchestrationClient)
+                self._add_rich_client(fb, client_name, DurableFunctionsClient)
 
                 fb.add_binding(
                     binding=DurableClient(name=client_name,
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/internal/DurableClientConverter.py b/durabletask-azurefunctions/durabletask/azurefunctions/internal/DurableClientConverter.py
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/internal/__init__.py b/durabletask-azurefunctions/durabletask/azurefunctions/internal/__init__.py
@@ -1,3 +0,0 @@
-from .DurableClientConverter import DurableClientConverter
-
-__all__ = ["DurableClientConverter"]
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/internal/azurefunctions_grpc_interceptor.py b/durabletask-azurefunctions/durabletask/azurefunctions/internal/azurefunctions_grpc_interceptor.py
@@ -0,0 +1,27 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from importlib.metadata import version
+
+from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
+
+
+class AzureFunctionsDefaultClientInterceptorImpl (DefaultClientInterceptorImpl):
+    """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
+    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an
+    interceptor to add additional headers to all calls as needed."""
+    required_query_string_parameters: str
+
+    def __init__(self, taskhub_name: str, required_query_string_parameters: str):
+        self.required_query_string_parameters = required_query_string_parameters
+        try:
+            # Get the version of the azurefunctions package
+            sdk_version = version('durabletask-azurefunctions')
+        except Exception:
+            # Fallback if version cannot be determined
+            sdk_version = "unknown"
+        user_agent = f"durabletask-python/{sdk_version}"
+        self._metadata = [
+            ("taskhub", taskhub_name),
+            ("x-user-agent", user_agent)]  # 'user-agent' is a reserved header in grpc, so we use 'x-user-agent' instead
+        super().__init__(self._metadata)
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/internal/azurefunctions_null_stub.py b/durabletask-azurefunctions/durabletask/azurefunctions/internal/azurefunctions_null_stub.py
@@ -0,0 +1,39 @@
+
+from durabletask.internal.ProtoTaskHubSidecarServiceStub import ProtoTaskHubSidecarServiceStub
+
+
+class AzureFunctionsNullStub(ProtoTaskHubSidecarServiceStub):
+    """Missing associated documentation comment in .proto file."""
+
+    def __init__(self):
+        """Constructor.
+
+        Args:
+            channel: A grpc.Channel.
+        """
+        self.Hello = lambda *args, **kwargs: None
+        self.StartInstance = lambda *args, **kwargs: None
+        self.GetInstance = lambda *args, **kwargs: None
+        self.RewindInstance = lambda *args, **kwargs: None
+        self.WaitForInstanceStart = lambda *args, **kwargs: None
+        self.WaitForInstanceCompletion = lambda *args, **kwargs: None
+        self.RaiseEvent = lambda *args, **kwargs: None
+        self.TerminateInstance = lambda *args, **kwargs: None
+        self.SuspendInstance = lambda *args, **kwargs: None
+        self.ResumeInstance = lambda *args, **kwargs: None
+        self.QueryInstances = lambda *args, **kwargs: None
+        self.PurgeInstances = lambda *args, **kwargs: None
+        self.GetWorkItems = lambda *args, **kwargs: None
+        self.CompleteActivityTask = lambda *args, **kwargs: None
+        self.CompleteOrchestratorTask = lambda *args, **kwargs: None
+        self.CompleteEntityTask = lambda *args, **kwargs: None
+        self.StreamInstanceHistory = lambda *args, **kwargs: None
+        self.CreateTaskHub = lambda *args, **kwargs: None
+        self.DeleteTaskHub = lambda *args, **kwargs: None
+        self.SignalEntity = lambda *args, **kwargs: None
+        self.GetEntity = lambda *args, **kwargs: None
+        self.QueryEntities = lambda *args, **kwargs: None
+        self.CleanEntityStorage = lambda *args, **kwargs: None
+        self.AbandonTaskActivityWorkItem = lambda *args, **kwargs: None
+        self.AbandonTaskOrchestratorWorkItem = lambda *args, **kwargs: None
+        self.AbandonTaskEntityWorkItem = lambda *args, **kwargs: None
diff --git a/durabletask-azurefunctions/durabletask/azurefunctions/worker.py b/durabletask-azurefunctions/durabletask/azurefunctions/worker.py
@@ -1,2 +1,32 @@
-class TempClass:
-    pass
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from threading import Event
+from durabletask.worker import _Registry, ConcurrencyOptions
+from durabletask.internal import shared
+from durabletask.worker import TaskHubGrpcWorker
+
+
+# Worker class used for Durable Task Scheduler (DTS)
+class DurableFunctionsWorker(TaskHubGrpcWorker):
+    """TOOD: Docs
+    """
+
+    def __init__(self):
+        # Don't call the parent constructor - we don't actually want to start an AsyncWorkerLoop
+        # or recieve work items from anywhere but the method that is creating this worker
+        self._registry = _Registry()
+        self._host_address = ""
+        self._logger = shared.get_logger("worker")
+        self._shutdown = Event()
+        self._is_running = False
+        self._secure_channel = False
+
+        self._concurrency_options = ConcurrencyOptions()
+
+        self._interceptors = None
+
+    def add_named_orchestrator(self, name: str, func):
+        """TOOD: Docs
+        """
+        self._registry.add_named_orchestrator(name, func)
diff --git a/durabletask/internal/ProtoTaskHubSidecarServiceStub.py b/durabletask/internal/ProtoTaskHubSidecarServiceStub.py
diff --git a/durabletask/worker.py b/durabletask/worker.py

Original file line number	Diff line number	Diff line change
`@@ -1,3 +0,0 @@`
`1`		`-from .DurableClientConverter import DurableClientConverter`
`2`		`-`
`3`		`-__all__ = ["DurableClientConverter"]`