Refine entity support

- Still needs eventSent and eventRecieved implementations

Author: andystaples

durabletask-azurefunctions/durabletask/azurefunctions/client.py

@@ -6,10 +6,12 @@
 from datetime import timedelta
 from typing import Any, Optional
 import azure.functions as func
+from urllib.parse import urlparse, quote
 
 from durabletask.entities import EntityInstanceId
 from durabletask.client import TaskHubGrpcClient
 from durabletask.azurefunctions.internal.azurefunctions_grpc_interceptor import AzureFunctionsDefaultClientInterceptorImpl
+from durabletask.azurefunctions.http import HttpManagementPayload
 
 
 # Client class used for Durable Functions
@@ -56,30 +58,32 @@ def create_check_status_response(self, request: func.HttpRequest, instance_id: s
             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.")
+        location_url = self._get_instance_status_url(request, instance_id)
+        return func.HttpResponse(
+            body=str(self._get_client_response_links(request, instance_id)),
+            status_code=501,
+            headers={
+                'content-type': 'application/json',
+                'Location': location_url,
+            },
+        )
 
-    def create_http_management_payload(self, instance_id: str) -> dict[str, str]:
+    def create_http_management_payload(self, request: func.HttpRequest, instance_id: str) -> HttpManagementPayload:
         """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.")
+        return self._get_client_response_links(request, instance_id)
 
-    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.
+    def _get_client_response_links(self, request: func.HttpRequest, instance_id: str) -> HttpManagementPayload:
+        instance_status_url = self._get_instance_status_url(request, instance_id)
+        return HttpManagementPayload(instance_id, instance_status_url, self.requiredQueryStringParameters)
 
-        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.")
+    @staticmethod
+    def _get_instance_status_url(request: func.HttpRequest, instance_id: str) -> str:
+        request_url = urlparse(request.url)
+        location_url = f"{request_url.scheme}://{request_url.netloc}{request_url.path}"
+        encoded_instance_id = quote(instance_id)
+        location_url = location_url + "/runtime/webhooks/durabletask/instances/" + encoded_instance_id
+        return location_url

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

@@ -96,7 +96,7 @@ def stub_complete(stub_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
+            handle.orchestrator_function = orchestrator_func  # type: ignore
 
             # invoke next decorator, with the Orchestrator as input
             handle.__name__ = orchestrator_func.__name__
@@ -131,15 +131,7 @@ def handle(context) -> str:
                     context_body = context
                 orchestration_context = context_body
                 request = EntityBatchRequest()
-                request_2 = EntityRequest()
-                try:
-                    request.ParseFromString(base64.b64decode(orchestration_context))
-                except Exception:
-                    pass
-                try:
-                    request_2.ParseFromString(base64.b64decode(orchestration_context))
-                except Exception:
-                    pass
+                request.ParseFromString(base64.b64decode(orchestration_context))
                 stub = AzureFunctionsNullStub()
                 worker = DurableFunctionsWorker()
                 response: Optional[EntityBatchResult] = None
@@ -157,7 +149,7 @@ def stub_complete(stub_response: EntityBatchResult):
                 # The Python worker returns the input as type "json", so double-encoding is necessary
                 return '"' + base64.b64encode(response.SerializeToString()).decode('utf-8') + '"'
 
-            handle.entity_function = entity_func
+            handle.entity_function = entity_func  # type: ignore
 
             # invoke next decorator, with the Entity as input
             handle.__name__ = entity_func.__name__

durabletask-azurefunctions/durabletask/azurefunctions/http/__init__.py

@@ -0,0 +1,6 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from durabletask.azurefunctions.http.http_management_payload import HttpManagementPayload
+
+__all__ = ["HttpManagementPayload"]

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

@@ -0,0 +1,18 @@
+import json
+
+
+class HttpManagementPayload:
+    def __init__(self, instance_id: str, instance_status_url: str, required_query_string_parameters: str):
+        self.urls = {
+            'id': instance_id,
+            'purgeHistoryDeleteUri': instance_status_url + "?" + required_query_string_parameters,
+            'restartPostUri': instance_status_url + "/restart?" + required_query_string_parameters,
+            'sendEventPostUri': instance_status_url + "/raiseEvent/{eventName}?" + required_query_string_parameters,
+            'statusQueryGetUri': instance_status_url + "?" + required_query_string_parameters,
+            'terminatePostUri': instance_status_url + "/terminate?reason={text}&" + required_query_string_parameters,
+            'resumePostUri': instance_status_url + "/resume?reason={text}&" + required_query_string_parameters,
+            'suspendPostUri': instance_status_url + "/suspend?reason={text}&" + required_query_string_parameters
+        }
+
+    def __str__(self):
+        return json.dumps(self.urls)

durabletask/internal/helpers.py

@@ -4,6 +4,7 @@
 import traceback
 from datetime import datetime
 from typing import Optional
+import uuid
 
 from google.protobuf import timestamp_pb2, wrappers_pb2
 
@@ -197,8 +198,9 @@ 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())
     return pb.OrchestratorAction(id=id, sendEntityMessage=pb.SendEntityMessageAction(entityOperationCalled=pb.EntityOperationCalledEvent(
-        requestId=f"{parent_instance_id}:{id}",
+        requestId=request_id,
         operation=operation,
         scheduledTime=None,
         input=get_string_value(encoded_input),
@@ -209,8 +211,9 @@ 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())
     return pb.OrchestratorAction(id=id, sendEntityMessage=pb.SendEntityMessageAction(entityOperationSignaled=pb.EntityOperationSignaledEvent(
-        requestId=f"{entity_id}:{id}",
+        requestId=request_id,
         operation=operation,
         scheduledTime=None,
         input=get_string_value(encoded_input),

durabletask/worker.py

@@ -13,6 +13,7 @@
 from types import GeneratorType
 from enum import Enum
 from typing import Any, Generator, Optional, Sequence, TypeVar, Union
+import uuid
 from packaging.version import InvalidVersion, parse
 
 import grpc
@@ -740,6 +741,7 @@ def _execute_entity_batch(
             stub: Union[stubs.TaskHubSidecarServiceStub, ProtoTaskHubSidecarServiceStub],
             completionToken,
     ):
+        operation_infos = None
         if isinstance(req, pb.EntityRequest):
             req, operation_infos = helpers.convert_to_entity_batch_request(req)
 
@@ -1200,7 +1202,7 @@ def lock_entities_function_helper(self, id: int, entities: list[EntityInstanceId
         if not transition_valid:
             raise RuntimeError(error_message)
 
-        critical_section_id = f"{self.instance_id}:{id:04x}"
+        critical_section_id = str(uuid.uuid4())
 
         request, target = self._entity_context.emit_acquire_message(critical_section_id, entities)
 
@@ -1747,6 +1749,12 @@ def process_event(
             elif event.HasField("orchestratorCompleted"):
                 # Added in Functions only (for some reason) and does not affect orchestrator flow
                 pass
+            elif event.HasField("eventSent"):
+                # Added in Functions only (for some reason) and does not affect orchestrator flow
+                pass
+            elif event.HasField("eventRaised"):
+                # Added in Functions only (for some reason) and does not affect orchestrator flow
+                pass
             else:
                 eventType = event.WhichOneof("eventType")
                 raise task.OrchestrationStateError(