Add instance_id_prefix support to get_status_by

Adds an optional instance_id_prefix parameter to RpcManagementOptions and DurableOrchestrationClient.get_status_by, emitting it as the instanceIdPrefix query string parameter on the management URL. This brings the Python SDK to parity with the .NET SDKs and the underlying Durable Task extension HTTP API.

Fixes #601

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: cgillum

azure/durable_functions/models/DurableOrchestrationClient.py

@@ -354,7 +354,8 @@ async def get_status_all(self) -> List[DurableOrchestrationStatus]:
 
     async def get_status_by(self, created_time_from: datetime = None,
                             created_time_to: datetime = None,
-                            runtime_status: List[OrchestrationRuntimeStatus] = None) \
+                            runtime_status: List[OrchestrationRuntimeStatus] = None,
+                            instance_id_prefix: str = None) \
             -> List[DurableOrchestrationStatus]:
         """Get the status of all orchestration instances that match the specified conditions.
 
@@ -367,6 +368,8 @@ async def get_status_by(self, created_time_from: datetime = None,
         runtime_status: List[OrchestrationRuntimeStatus]
             Return orchestration instances which match any of the runtimeStatus values
             in this list.
+        instance_id_prefix: str
+            Return orchestration instances whose instance ID starts with this prefix.
 
         Returns
         -------
@@ -376,7 +379,8 @@ async def get_status_by(self, created_time_from: datetime = None,
         # TODO: do we really want folks to us this without specifying all the args?
         options = RpcManagementOptions(created_time_from=created_time_from,
                                        created_time_to=created_time_to,
-                                       runtime_status=runtime_status)
+                                       runtime_status=runtime_status,
+                                       instance_id_prefix=instance_id_prefix)
         request_url = options.to_url(self._orchestration_bindings.rpc_base_url)
         response = await self._get_async_request(request_url)
         switch_statement = {

azure/durable_functions/models/RpcManagementOptions.py

@@ -16,7 +16,8 @@ def __init__(self, instance_id: str = None, task_hub_name: str = None,
                  created_time_to: datetime = None,
                  runtime_status: List[OrchestrationRuntimeStatus] = None, show_input: bool = None,
                  operation_name: str = None,
-                 entity_Id: EntityId = None):
+                 entity_Id: EntityId = None,
+                 instance_id_prefix: str = None):
         self._instance_id = instance_id
         self._task_hub_name = task_hub_name
         self._connection_name = connection_name
@@ -28,6 +29,7 @@ def __init__(self, instance_id: str = None, task_hub_name: str = None,
         self._show_input = show_input
         self.operation_name = operation_name
         self.entity_Id = entity_Id
+        self._instance_id_prefix = instance_id_prefix
 
     @staticmethod
     def _add_arg(query: List[str], name: str, value: Any):
@@ -76,6 +78,7 @@ def to_url(self, base_url: Optional[str]) -> str:
         self._add_date_arg(query, 'createdTimeFrom', self._created_time_from)
         self._add_date_arg(query, 'createdTimeTo', self._created_time_to)
         self._add_arg(query, 'op', self.operation_name)
+        self._add_arg(query, 'instanceIdPrefix', self._instance_id_prefix)
         if self._runtime_status is not None and len(self._runtime_status) > 0:
             runtime_status = ",".join(r.value for r in self._runtime_status)
             self._add_arg(query, 'runtimeStatus', runtime_status)

tests/models/test_DurableOrchestrationClient.py

@@ -285,6 +285,21 @@ async def test_get_500_get_status_by_failed(binding_string):
         await client.get_status_by(runtime_status=[OrchestrationRuntimeStatus.Running])
 
 
+@pytest.mark.asyncio
+async def test_get_200_get_status_by_with_instance_id_prefix(binding_string):
+    mock_request = MockRequest(
+        expected_url=f"{RPC_BASE_URL}instances/?instanceIdPrefix=940c5f519eb0",
+        response=[200, [dict(createdTime=TEST_CREATED_TIME,
+                             lastUpdatedTime=TEST_LAST_UPDATED_TIME,
+                             runtimeStatus="Running")]])
+    client = DurableOrchestrationClient(binding_string)
+    client._get_async_request = mock_request.get
+
+    result = await client.get_status_by(instance_id_prefix="940c5f519eb0")
+    assert result is not None
+    assert len(result) == 1
+
+
 @pytest.mark.asyncio
 async def test_get_200_get_status_all_success(binding_string):
     mock_request = MockRequest(expected_url=f"{RPC_BASE_URL}instances/",

tests/models/test_RpcManagementOptions.py

@@ -75,5 +75,12 @@ def test_datetime_status():
     to_as_string = created_time_to.strftime(DATETIME_STRING_FORMAT)
     expected = f"{RPC_BASE_URL}instances/?createdTimeFrom={from_as_string}" \
                f"&createdTimeTo={to_as_string}"
+    assert_urls_match(expected=expected, result=result)
+
+
+def test_instance_id_prefix():
+    options = RpcManagementOptions(instance_id_prefix='940c5f519eb0')
+    result = options.to_url(RPC_BASE_URL)
+    expected = f"{RPC_BASE_URL}instances/?instanceIdPrefix=940c5f519eb0"
 
     assert_urls_match(expected=expected, result=result)