Complete durable entities implementation with examples and documentation

Co-authored-by: berndverst <[email protected]>

Author: Copilot

README.md

@@ -117,6 +117,38 @@ Orchestrations can start child orchestrations using the `call_sub_orchestrator`
 
 Orchestrations can wait for external events using the `wait_for_external_event` API. External events are useful for implementing human interaction patterns, such as waiting for a user to approve an order before continuing.
 
+### Durable entities
+
+Durable entities are stateful objects that can maintain state across multiple operations. Entities support operations that can read and modify the entity's state. Each entity has a unique entity ID and maintains its state independently.
+
+```python
+# Define an entity function
+def counter_entity(ctx: task.EntityContext, input):
+    if ctx.operation_name == "increment":
+        current_count = ctx.get_state() or 0
+        new_count = current_count + (input or 1)
+        ctx.set_state(new_count)
+        return new_count
+    elif ctx.operation_name == "get":
+        return ctx.get_state() or 0
+
+# Register the entity with the worker
+worker.add_named_entity("Counter", counter_entity)
+
+# Signal an entity from an orchestrator
+yield ctx.signal_entity("Counter@my-counter", "increment", input=5)
+
+# Or signal an entity directly from a client
+client.signal_entity("Counter@my-counter", "increment", input=10)
+
+# Query entity state
+entity_state = client.get_entity("Counter@my-counter", include_state=True)
+if entity_state and entity_state.exists:
+    print(f"Current count: {entity_state.serialized_state}")
+```
+
+You can find the full sample [here](./examples/durable_entities.py).
+
 ### Continue-as-new (TODO)
 
 Orchestrations can be continued as new using the `continue_as_new` API. This API allows an orchestration to restart itself from scratch, optionally with a new input.

durabletask/client.py

@@ -269,7 +269,7 @@ def get_entity(self, entity_id: str, *, include_state: bool = True) -> Optional[
         """
         req = pb.GetEntityRequest(instanceId=entity_id, includeState=include_state)
         res: pb.GetEntityResponse = self._stub.GetEntity(req)
-        
+
         if not res.exists:
             return None
 
@@ -357,6 +357,6 @@ def clean_entity_storage(self, *,
         self._logger.info("Cleaning entity storage.")
         res: pb.CleanEntityStorageResponse = self._stub.CleanEntityStorage(req)
 
-        return (res.emptyEntitiesRemoved, 
+        return (res.emptyEntitiesRemoved,
                 res.orphanedLocksReleased,
                 res.continuationToken.value if not helpers.is_empty(res.continuationToken) else None)

durabletask/worker.py

@@ -940,15 +940,15 @@ def signal_entity(self, entity_id: str, operation_name: str, *,
 
         # Entity signals don't return values, so we create a completed task
         signal_task = task.CompletableTask()
-        
+
         # Store the action to be executed
         task_id = self._next_task_id()
         self._pending_actions[task_id] = action
         self._pending_tasks[task_id] = signal_task
-        
+
         # Mark as complete since signals don't have return values
         signal_task.complete(None)
-        
+
         return signal_task
 
     def call_entity(self, entity_id: str, operation_name: str, *,
@@ -1372,15 +1372,15 @@ def execute(self, req: pb.EntityBatchRequest) -> pb.EntityBatchResult:
 
         # Parse current entity state
         current_state = shared.from_json(req.entityState.value) if not ph.is_empty(req.entityState) else None
-        
+
         # Extract entity type from instance ID (format: entitytype@key)
         entity_type = "Unknown"
         if "@" in instance_id:
             entity_type = instance_id.split("@")[0]
-        
+
         results = []
         actions = []
-        
+
         for operation in req.operations:
             try:
                 # Get the entity function using the entity type from instanceId
@@ -1413,12 +1413,12 @@ def execute(self, req: pb.EntityBatchRequest) -> pb.EntityBatchResult:
                     ))
                 else:
                     result.success.CopyFrom(pb.OperationResultSuccess())
-                
+
                 results.append(result)
 
             except Exception as ex:
                 self._logger.exception(f"Error executing entity operation '{operation.operation}' on entity type '{entity_type}': {ex}")
-                
+
                 # Create failure result
                 failure_details = ph.new_failure_details(ex)
                 result = pb.OperationResult()

examples/durable_entities.py

@@ -0,0 +1,146 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+Example demonstrating durable entities usage.
+
+This example shows how to create and use durable entities with the Python SDK.
+Entities are stateful objects that can maintain state across multiple operations.
+"""
+
+import durabletask.task as dt
+from durabletask.worker import TaskHubGrpcWorker
+import logging
+
+
+def counter_entity(ctx: dt.EntityContext, input) -> int:
+    """A simple counter entity that can increment, decrement, get, and reset."""
+
+    if ctx.operation_name == "increment":
+        current_count = ctx.get_state() or 0
+        increment_by = input or 1
+        new_count = current_count + increment_by
+        ctx.set_state(new_count)
+        return new_count
+
+    elif ctx.operation_name == "decrement":
+        current_count = ctx.get_state() or 0
+        decrement_by = input or 1
+        new_count = current_count - decrement_by
+        ctx.set_state(new_count)
+        return new_count
+
+    elif ctx.operation_name == "get":
+        return ctx.get_state() or 0
+
+    elif ctx.operation_name == "reset":
+        ctx.set_state(0)
+        return 0
+
+    else:
+        raise ValueError(f"Unknown operation: {ctx.operation_name}")
+
+
+def shopping_cart_entity(ctx: dt.EntityContext, input):
+    """A shopping cart entity that can add/remove items and calculate totals."""
+
+    if ctx.operation_name == "add_item":
+        cart = ctx.get_state() or {"items": []}
+        cart["items"].append(input)
+        ctx.set_state(cart)
+        return len(cart["items"])
+
+    elif ctx.operation_name == "remove_item":
+        cart = ctx.get_state() or {"items": []}
+        if input in cart["items"]:
+            cart["items"].remove(input)
+            ctx.set_state(cart)
+        return len(cart["items"])
+
+    elif ctx.operation_name == "get_items":
+        cart = ctx.get_state() or {"items": []}
+        return cart["items"]
+
+    elif ctx.operation_name == "get_total":
+        cart = ctx.get_state() or {"items": []}
+        # Simple total calculation assuming each item has a 'price' field
+        total = sum(item.get("price", 0) for item in cart["items"] if isinstance(item, dict))
+        return total
+
+    elif ctx.operation_name == "clear":
+        ctx.set_state({"items": []})
+        return 0
+
+    else:
+        raise ValueError(f"Unknown operation: {ctx.operation_name}")
+
+
+def entity_orchestrator(ctx: dt.OrchestrationContext, input):
+    """Orchestrator that demonstrates entity interactions."""
+
+    # Signal entities (fire-and-forget)
+    yield ctx.signal_entity("Counter@global", "increment", input=5)
+    yield ctx.signal_entity("Counter@user1", "increment", input=1)
+    yield ctx.signal_entity("Counter@user2", "increment", input=2)
+
+    # Add items to shopping cart
+    yield ctx.signal_entity("ShoppingCart@user1", "add_item",
+                          input={"name": "Apple", "price": 1.50})
+    yield ctx.signal_entity("ShoppingCart@user1", "add_item",
+                          input={"name": "Banana", "price": 0.75})
+
+    return "Entity operations completed"
+
+
+def main():
+    # Set up logging
+    logging.basicConfig(level=logging.INFO)
+
+    # Create and configure the worker
+    worker = TaskHubGrpcWorker()
+
+    # Register entities - entities should be registered by their intended name
+    # Since entity execution extracts the entity type from the instance ID (e.g., "Counter@key1")
+    # we need to register them with the exact name that will be used in instance IDs
+    worker._registry.add_named_entity("Counter", counter_entity)
+    worker._registry.add_named_entity("ShoppingCart", shopping_cart_entity)
+
+    # Register orchestrator
+    worker.add_orchestrator(entity_orchestrator)
+
+    print("Entity worker example setup complete.")
+    print("\nRegistered entities:")
+    print("- Counter: supports increment, decrement, get, reset operations")
+    print("- ShoppingCart: supports add_item, remove_item, get_items, get_total, clear operations")
+    print("\nTo use entities, you would:")
+    print("1. Start the worker: worker.start()")
+    print("2. Use a client to signal entities or start orchestrations")
+    print("3. Query entity state using client.get_entity()")
+
+    # Example client usage (commented out since it requires a running sidecar)
+    """
+    # Create client
+    client = TaskHubGrpcClient()
+
+    # Start an orchestration that uses entities
+    instance_id = client.schedule_new_orchestration(entity_orchestrator)
+    print(f"Started orchestration: {instance_id}")
+
+    # Signal entities directly
+    client.signal_entity("Counter@test", "increment", input=10)
+    client.signal_entity("Counter@test", "increment", input=5)
+
+    # Query entity state
+    counter_state = client.get_entity("Counter@test", include_state=True)
+    if counter_state:
+        print(f"Counter state: {counter_state.serialized_state}")
+
+    # Query entities
+    query = dt.EntityQuery(instance_id_starts_with="Counter@", include_state=True)
+    results = client.query_entities(query)
+    print(f"Found {len(results.entities)} counter entities")
+    """
+
+
+if __name__ == "__main__":
+    main()