diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index e9c022e..804ae4d 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -32,6 +32,9 @@ avoiding repeated allocation of unused worker resources. FIXED +- The v1-compatible `get_status()` API now supports `show_history` and + `show_history_output`, including compacted `historyEvents` output without an + additional history request when history is not requested. - HTTP management payloads now preserve the host-provided management URL templates and configured HTTP base paths, include `rewindPostUri`, encode instance IDs, and use forwarded request origins when enabled by the host. @@ -206,7 +209,3 @@ code: - Orchestration history is not exposed on the context; `DurableOrchestrationContext.histories` raises `NotImplementedError`. Use the client's `get_orchestration_history(...)` instead. -- The client status methods accept the v1 `show_history` / - `show_history_output` flags for signature compatibility but ignore them, so - the returned status has no `historyEvents`. Use - `get_orchestration_history(...)` to retrieve history. diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md index e43165f..3a71875 100644 --- a/azure-functions-durable/MIGRATION_GUIDE.md +++ b/azure-functions-durable/MIGRATION_GUIDE.md @@ -269,8 +269,6 @@ instead of calling `context.set_result`. - `DurableOrchestrationContext.histories` is unavailable. Retrieve history with `client.get_orchestration_history(instance_id)`. -- The compatibility `show_history` and `show_history_output` status flags are - accepted but ignored. - Distributed tracing is not yet wired through the Python provider. - Continuous history export is not supported by Azure Functions. - Unusual callable signatures can be misclassified by the compatibility layer. diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index aef0633..512b6f5 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -28,6 +28,7 @@ from .http.http_management_payload import HttpManagementPayload, replace_url_origin from .internal.compat.durable_orchestration_status import DurableOrchestrationStatus from .internal.compat.entity_state_response import EntityStateResponse +from .internal.compat.history_projection import project_history from .internal.compat.orchestration_runtime_status import OrchestrationRuntimeStatus, to_durabletask_statuses from .internal.compat.purge_history_result import PurgeHistoryResult @@ -339,15 +340,27 @@ async def get_status( ``OrchestrationState`` for v1 back-compat. When the instance does not exist, a falsy status is returned rather than ``None``. - The ``show_history`` and ``show_history_output`` flags have no - equivalent in durabletask and are ignored. Payloads are fetched to + When ``show_history`` is true, history is fetched and projected into + the v1 status-query shape. ``show_history_output`` controls whether + output-bearing history fields are included. Payloads are fetched to preserve the v1 output, custom-status, and failure-detail fields; ``show_input`` controls whether the compatibility wrapper exposes the - input. + orchestration and history inputs. """ state = await self.get_orchestration_state(instance_id, fetch_payloads=True) + projected_history = None + if (show_history + and state is not None + and state.runtime_status != OrchestrationStatus.PENDING): + history = await self.get_orchestration_history(instance_id) + projected_history = project_history( + history, + show_input=show_input, + show_history_output=show_history_output) return DurableOrchestrationStatus.from_orchestration_state( - state, include_input=show_input) + state, + include_input=show_input, + history=projected_history) @deprecated("get_status_all is deprecated; use get_all_orchestration_states instead.") async def get_status_all(self) -> list[DurableOrchestrationStatus]: diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py index cc6ea3e..17f6b2b 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py @@ -27,18 +27,23 @@ class DurableOrchestrationStatus: the v1 behaviour where ``get_status`` never returned ``None``. """ - def __init__(self, state: Optional[OrchestrationState] = None): + def __init__( + self, + state: Optional[OrchestrationState] = None, + history: Optional[list[Any]] = None): self._state = state self._include_input = True + self._history = history @classmethod def from_orchestration_state( cls, state: Optional[OrchestrationState], *, - include_input: bool = True) -> "DurableOrchestrationStatus": + include_input: bool = True, + history: Optional[list[Any]] = None) -> "DurableOrchestrationStatus": """Wrap a durabletask ``OrchestrationState`` (or ``None``).""" - status = cls(state) + status = cls(state, history) status._include_input = include_input return status @@ -77,7 +82,8 @@ def _reserialize(value: Any) -> Optional[str]: serialized_custom_status=_reserialize(data.get("customStatus")), failure_details=None, ) - return cls(state) + history = data.get("historyEvents", data.get("history")) + return cls(state, history) def __bool__(self) -> bool: return self._state is not None @@ -145,10 +151,10 @@ def custom_status(self) -> Any: def history(self) -> Optional[list[Any]]: """Get the execution history. - History is not available through this compatibility path and is always - ``None``; use ``get_orchestration_history`` on the client instead. + The history is populated only when requested by passing + ``show_history=True`` to the client's compatibility ``get_status`` API. """ - return None + return self._history def to_json(self) -> dict[str, Any]: """Convert this status into a v1-compatible JSON dictionary. @@ -179,6 +185,8 @@ def to_json(self) -> dict[str, Any]: self._state.serialized_custom_status if self._state is not None else None) if custom_status is not None: result["customStatus"] = custom_status + if self.history is not None: + result["historyEvents"] = self.history return result @staticmethod diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py new file mode 100644 index 0000000..fcf6ea1 --- /dev/null +++ b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py @@ -0,0 +1,189 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import json +from dataclasses import fields, is_dataclass +from datetime import datetime, timezone +from typing import Any, cast + +from durabletask import history, task +from durabletask.client import OrchestrationStatus + +from .orchestration_runtime_status import from_durabletask_status + + +def project_history( + events: list[history.HistoryEvent], + *, + show_input: bool, + show_history_output: bool) -> list[dict[str, Any]]: + """Project Durable Task history events into the v1 status-query shape.""" + projected: list[dict[str, Any]] = [] + scheduled_tasks: dict[int, tuple[int, history.TaskScheduledEvent]] = {} + scheduled_sub_orchestrations: dict[ + int, tuple[int, history.SubOrchestrationInstanceCreatedEvent]] = {} + removed_indexes: set[int] = set() + + for event in events: + if isinstance( + event, + (history.OrchestratorStartedEvent, + history.OrchestratorCompletedEvent)): + continue + + item = _project_event( + event, + show_input=show_input, + show_history_output=show_history_output) + projected_index = len(projected) + + if isinstance(event, history.TaskScheduledEvent): + scheduled_tasks[event.event_id] = (projected_index, event) + elif isinstance( + event, + (history.TaskCompletedEvent, history.TaskFailedEvent)): + _add_scheduled_event_data( + item, + event.task_scheduled_id, + scheduled_tasks, + removed_indexes, + show_input) + elif isinstance(event, history.SubOrchestrationInstanceCreatedEvent): + scheduled_sub_orchestrations[event.event_id] = ( + projected_index, event) + elif isinstance( + event, + (history.SubOrchestrationInstanceCompletedEvent, + history.SubOrchestrationInstanceFailedEvent)): + _add_scheduled_event_data( + item, + event.task_scheduled_id, + scheduled_sub_orchestrations, + removed_indexes, + show_input) + + projected.append(item) + + return [ + event for index, event in enumerate(projected) + if index not in removed_indexes + ] + + +def _project_event( + event: history.HistoryEvent, + *, + show_input: bool, + show_history_output: bool) -> dict[str, Any]: + item = { + "EventType": type(event).__name__.removesuffix("Event"), + "Timestamp": _format_datetime(event.timestamp), + } + for field in fields(event): + if field.name in {"event_id", "timestamp"}: + continue + if field.name == "input" and not show_input: + continue + if field.name in {"result", "output"} and not show_history_output: + continue + + value = getattr(event, field.name) + if value is None: + continue + if field.name in {"result", "output"}: + value = _raw_payload(value) + item[_field_name(field.name)] = _serialize(value) + + if isinstance( + event, + (history.TaskScheduledEvent, + history.SubOrchestrationInstanceCreatedEvent)): + item.pop("Version", None) + elif isinstance(event, history.ExecutionStartedEvent): + item["FunctionName"] = item.pop("Name") + for field_name in ( + "OrchestrationInstance", + "ParentInstance", + "Version", + "Tags"): + item.pop(field_name, None) + elif isinstance( + event, + (history.TaskCompletedEvent, + history.TaskFailedEvent, + history.SubOrchestrationInstanceCompletedEvent, + history.SubOrchestrationInstanceFailedEvent)): + item.pop("TaskScheduledId", None) + elif isinstance(event, history.ExecutionCompletedEvent): + try: + status = OrchestrationStatus(event.orchestration_status) + except ValueError: + pass + else: + item["OrchestrationStatus"] = from_durabletask_status(status).name + elif isinstance(event, history.TimerFiredEvent): + item.pop("TimerId", None) + + return item + + +def _add_scheduled_event_data( + item: dict[str, Any], + scheduled_id: int, + scheduled_events: dict[int, tuple[int, Any]], + removed_indexes: set[int], + show_input: bool) -> None: + scheduled = scheduled_events.get(scheduled_id) + if scheduled is None: + return + + scheduled_index, scheduled_event = scheduled + item["ScheduledTime"] = _format_datetime(scheduled_event.timestamp) + item["FunctionName"] = scheduled_event.name + if show_input and scheduled_event.input is not None: + item["Input"] = scheduled_event.input + removed_indexes.add(scheduled_index) + + +def _serialize(value: Any) -> Any: + if isinstance(value, datetime): + return _format_datetime(value) + if isinstance(value, task.FailureDetails): + result = { + "ErrorMessage": value.message, + "ErrorType": value.error_type, + } + if value.stack_trace is not None: + result["StackTrace"] = value.stack_trace + return result + if is_dataclass(value) and not isinstance(value, type): + return { + _field_name(field.name): _serialize(getattr(value, field.name)) + for field in fields(value) + if getattr(value, field.name) is not None + } + if isinstance(value, list): + return [_serialize(item) for item in cast(list[Any], value)] + if isinstance(value, dict): + mapping = cast(dict[Any, Any], value) + return {key: _serialize(item) for key, item in mapping.items()} + return value + + +def _field_name(name: str) -> str: + if name == "scheduled_start_timestamp": + return "ScheduledStartTime" + return "".join(part.title() for part in name.split("_")) + + +def _raw_payload(serialized: str) -> Any: + try: + return json.loads(serialized) + except (TypeError, ValueError): + return serialized + + +def _format_datetime(value: datetime) -> str: + if value.tzinfo is not None: + value = value.astimezone(timezone.utc).replace(tzinfo=None) + return value.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index fd48bd2..060f8a2 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -19,6 +19,7 @@ from azure.durable_functions.http.http_management_payload import ( replace_url_origin, ) +from durabletask import history as dt_history from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus from durabletask.entities import EntityInstanceId from durabletask.task import RetryPolicy @@ -738,10 +739,14 @@ async def test_get_status_suppresses_only_input_by_default(): client = _make_client() try: state = _fake_state(runtime_status=OrchestrationStatus.COMPLETED) + history_mock = AsyncMock() with patch.object(client, "get_orchestration_state", - new=AsyncMock(return_value=state)): + new=AsyncMock(return_value=state)), \ + patch.object(client, "get_orchestration_history", + new=history_mock): with pytest.warns(DeprecationWarning): status = await client.get_status("abc") + history_mock.assert_not_awaited() assert bool(status) is True assert status.name == "orch" assert status.instance_id == "abc" @@ -796,10 +801,14 @@ async def test_get_status_failed_preserves_failure_output(): async def test_get_status_missing_instance_is_falsy(): client = _make_client() try: + history_mock = AsyncMock() with patch.object(client, "get_orchestration_state", - new=AsyncMock(return_value=None)): + new=AsyncMock(return_value=None)), \ + patch.object(client, "get_orchestration_history", + new=history_mock): with pytest.warns(DeprecationWarning): - status = await client.get_status("missing") + status = await client.get_status("missing", show_history=True) + history_mock.assert_not_awaited() assert bool(status) is False assert status.runtime_status is None assert status.output is None @@ -807,6 +816,163 @@ async def test_get_status_missing_instance_is_falsy(): await client.close() +def _fake_history(): + started_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + return [ + dt_history.OrchestratorStartedEvent( + event_id=-1, timestamp=started_at), + dt_history.ExecutionStartedEvent( + event_id=0, + timestamp=started_at, + name="orch", + version="v1", + input='{"root": true}', + scheduled_start_timestamp=started_at + timedelta(minutes=1), + parent_trace_context=dt_history.TraceContext( + trace_parent="trace-parent", + span_id="span-id"), + orchestration_span_id="orchestration-span-id"), + dt_history.TaskScheduledEvent( + event_id=1, + timestamp=started_at + timedelta(seconds=1), + name="activity", + version="v1", + input='{"value": 1}'), + dt_history.OrchestratorCompletedEvent( + event_id=-1, + timestamp=started_at + timedelta(seconds=2)), + dt_history.TaskCompletedEvent( + event_id=2, + timestamp=started_at + timedelta(seconds=3), + task_scheduled_id=1, + result='{"result": 2}'), + dt_history.ExecutionCompletedEvent( + event_id=3, + timestamp=started_at + timedelta(seconds=4), + orchestration_status=OrchestrationStatus.COMPLETED.value, + result='{"done": true}'), + ] + + +@pytest.mark.parametrize("show_history_output", [False, True]) +async def test_get_status_projects_v1_history(show_history_output): + client = _make_client() + try: + history_mock = AsyncMock(return_value=_fake_history()) + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state())), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status( + "abc", + show_history=True, + show_history_output=show_history_output, + show_input=True) + + history_mock.assert_awaited_once_with("abc") + assert status.history is not None + assert [event["EventType"] for event in status.history] == [ + "ExecutionStarted", + "TaskCompleted", + "ExecutionCompleted", + ] + assert status.history[0]["FunctionName"] == "orch" + assert status.history[0]["Input"] == '{"root": true}' + assert status.history[0]["ScheduledStartTime"] == ( + "2026-01-01T00:01:00.000000Z") + assert "ScheduledStartTimestamp" not in status.history[0] + assert status.history[0]["ParentTraceContext"]["SpanId"] == "span-id" + assert status.history[0]["OrchestrationSpanId"] == ( + "orchestration-span-id") + assert status.history[1]["FunctionName"] == "activity" + assert status.history[1]["Input"] == '{"value": 1}' + assert status.history[1]["ScheduledTime"] == ( + "2026-01-01T00:00:01.000000Z") + assert status.history[2]["OrchestrationStatus"] == "Completed" + if show_history_output: + assert status.history[1]["Result"] == {"result": 2} + assert status.history[2]["Result"] == {"done": True} + else: + assert "Result" not in status.history[1] + assert "Result" not in status.history[2] + assert status.to_json()["historyEvents"] == status.history + finally: + await client.close() + + +async def test_get_status_history_suppresses_inputs(): + client = _make_client() + try: + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state())), \ + patch.object( + client, + "get_orchestration_history", + new=AsyncMock(return_value=_fake_history())): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + assert status.history is not None + assert all("Input" not in event for event in status.history) + finally: + await client.close() + + +@pytest.mark.parametrize( + "runtime_status", + [OrchestrationStatus.PENDING, None], +) +async def test_get_status_show_history_skips_unavailable_history(runtime_status): + client = _make_client() + try: + state = None if runtime_status is None else _fake_state(runtime_status) + history_mock = AsyncMock() + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=state)), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + history_mock.assert_not_awaited() + assert status.history is None + finally: + await client.close() + + +async def test_get_status_show_history_fetches_continued_as_new_history(): + client = _make_client() + try: + history_mock = AsyncMock(return_value=[]) + with patch.object( + client, + "get_orchestration_state", + new=AsyncMock(return_value=_fake_state( + OrchestrationStatus.CONTINUED_AS_NEW))), \ + patch.object( + client, + "get_orchestration_history", + new=history_mock): + with pytest.warns(DeprecationWarning): + status = await client.get_status("abc", show_history=True) + + history_mock.assert_awaited_once_with("abc") + assert status.history == [] + finally: + await client.close() + + async def test_get_status_all_returns_wrapped_list(): client = _make_client() try: diff --git a/tests/azure-functions-durable/test_durable_orchestration_status_compat.py b/tests/azure-functions-durable/test_durable_orchestration_status_compat.py index 3ba3c15..15f3681 100644 --- a/tests/azure-functions-durable/test_durable_orchestration_status_compat.py +++ b/tests/azure-functions-durable/test_durable_orchestration_status_compat.py @@ -88,9 +88,26 @@ def test_empty_status_is_falsy_with_none_attributes(): assert status.to_json() == {} -def test_history_is_always_none(): - status = DurableOrchestrationStatus.from_json(_sample_json()) - assert status.history is None +def test_history_events_round_trip(): + source = _sample_json() + source["historyEvents"] = [ + { + "EventType": "ExecutionCompleted", + "Result": {"done": True}, + }, + ] + status = DurableOrchestrationStatus.from_json(source) + assert status.history == source["historyEvents"] + assert status.to_json() == source + + +def test_legacy_history_field_is_accepted(): + status = DurableOrchestrationStatus.from_json({ + **_sample_json(), + "history": [{"EventType": "ExecutionStarted"}], + }) + assert status.history == [{"EventType": "ExecutionStarted"}] + assert status.to_json()["historyEvents"] == status.history # ---------------------------------------------------------------------------