Added read_entity_state to DurableOrchestrationClient (#285)

Author: martijnlentink

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -9,6 +9,7 @@
 
 from .PurgeHistoryResult import PurgeHistoryResult
 from .DurableOrchestrationStatus import DurableOrchestrationStatus
+from .EntityStateResponse import EntityStateResponse
 from .RpcManagementOptions import RpcManagementOptions
 from .OrchestrationRuntimeStatus import OrchestrationRuntimeStatus
 from ..models.DurableOrchestrationBindings import DurableOrchestrationBindings
@@ -132,6 +133,56 @@ def create_http_management_payload(self, instance_id: str) -> Dict[str, str]:
         """
         return self.get_client_response_links(None, instance_id)
 
+    async def read_entity_state(
+        self,
+        entityId: EntityId,
+        task_hub_name: Optional[str] = None,
+        connection_name: Optional[str] = None,
+    ) -> EntityStateResponse:
+        """Read the state of the entity.
+
+        Parameters
+        ----------
+        entityId : EntityId
+            The EntityId of the targeted entity.
+        task_hub_name : Optional[str]
+            The task hub name of the target entity.
+        connection_name : Optional[str]
+            The name of the connection string associated with [task_hub_name].
+
+        Raises
+        ------
+        Exception:
+            When an unexpected status code is returned
+
+        Returns
+        -------
+        EntityStateResponse
+            container object representing the state of the entity
+        """
+        options = RpcManagementOptions(
+            connection_name=connection_name,
+            task_hub_name=task_hub_name,
+            entity_Id=entityId,
+        )
+
+        request_url = options.to_url(self._orchestration_bindings.rpc_base_url)
+        response = await self._get_async_request(request_url)
+
+        switch_statement = {
+            200: lambda: EntityStateResponse(True, response[1]),
+            404: lambda: EntityStateResponse(False),
+        }
+
+        result = switch_statement.get(response[0])
+
+        if not result:
+            raise Exception(
+                f"The operation failed with an unexpected status code {response[0]}"
+            )
+
+        return result()
+
     def get_client_response_links(
             self,
             request: Optional[func.HttpRequest], instance_id: str) -> Dict[str, str]:

azure/durable_functions/models/EntityStateResponse.py

@@ -0,0 +1,23 @@
+from typing import Any
+
+
+class EntityStateResponse:
+    """Entity state response object for [read_entity_state]."""
+
+    def __init__(self, entity_exists: bool, entity_state: Any = None) -> None:
+        self._entity_exists = entity_exists
+        self._entity_state = entity_state
+
+    @property
+    def entity_exists(self) -> bool:
+        """Get the bool representing whether entity exists."""
+        return self._entity_exists
+
+    @property
+    def entity_state(self) -> Any:
+        """Get the state of the entity.
+
+        When [entity_exists] is False, this value will be None.
+        Optional.
+        """
+        return self._entity_state

samples/counter_entity/README.md

@@ -32,4 +32,15 @@ Http Functions:
 
 This indicates that your `DurableTrigger` function can be reached via a `GET` or `POST` request to that URL. `DurableTrigger` starts the function-chaning orchestrator whose name is passed as a parameter to the URL. So, to start the orchestrator, which is named `DurableOrchestration`, make a GET request to `http://127.0.0.1:7071/api/orchestrators/DurableOrchestration`.
 
-And that's it! You should see a JSON response with five URLs to monitor the status of the orchestration.
+And that's it! You should see a JSON response with five URLs to monitor the status of the orchestration.
+
+### Retrieving the state via the DurableOrchestrationClient
+It is possible to retrieve the state of an entity using the `read_entity_state` function. As an example we have the `RetrieveEntity` endpoint which will return the current state of the entity:
+
+```bash
+Http Functions:
+
+        RetrieveEntity: [GET] http://localhost:7071/api/entity/{entityName}/{entityKey}
+```
+
+For our example a call to `http://localhost:7071/api/entity/Counter/myCounter` will return the current state of our counter.

samples/counter_entity/RetrieveEntity/__init__.py

@@ -0,0 +1,26 @@
+import json
+import logging
+from typing import Any, Dict, Union, cast
+
+import azure.functions as func
+from azure.durable_functions import DurableOrchestrationClient
+from azure.durable_functions.models.utils.entity_utils import EntityId
+
+
+async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
+    client = DurableOrchestrationClient(starter)
+    entity_name, entity_key = req.route_params["entityName"], req.route_params["entityKey"]
+    
+    entity_identifier = EntityId(entity_name, entity_key)
+
+    entity_state_response = await client.read_entity_state(entity_identifier)
+
+    if not entity_state_response.entity_exists:
+        return func.HttpResponse("Entity not found", status_code=404)
+
+    return func.HttpResponse(json.dumps({
+        "entity": entity_name,
+        "key": entity_key,
+        "state": entity_state_response.entity_state
+    }))
+    

samples/counter_entity/RetrieveEntity/function.json

@@ -0,0 +1,25 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "authLevel": "function",
+      "name": "req",
+      "type": "httpTrigger",
+      "direction": "in",
+      "route": "entity/{entityName}/{entityKey}",
+      "methods": [
+        "get"
+      ]
+    },
+    {
+      "name": "$return",
+      "type": "http",
+      "direction": "out"
+    },
+    {
+      "name": "starter",
+      "type": "durableClient",
+      "direction": "in"
+    }
+  ]
+}