From 69872f11c55cced7ed193b125134a14cc5cbaa48 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Wed, 11 Nov 2026 13:28:29 -0700 Subject: [PATCH 1/2] Add Functions coroutine orchestrator support Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../azure/durable_functions/__init__.py | 6 +- .../decorators/durable_app.py | 7 +- .../internal/compat/orchestration_context.py | 290 +++++++- .../testing/orchestrator_generator_wrapper.py | 40 +- .../azure/durable_functions/worker.py | 12 +- .../e2e/apps/dtask_style/orchestrators.py | 22 + .../e2e/apps/v1_style/orchestrators.py | 23 + .../e2e/test_coroutine_e2e.py | 25 + .../test_async_compat_context.py | 656 ++++++++++++++++++ .../test_functions_coroutine_orchestrators.py | 222 ++++++ .../test_http_retry.py | 14 +- .../typing/import_package.py | 48 +- 12 files changed, 1335 insertions(+), 30 deletions(-) create mode 100644 tests/azure-functions-durable/e2e/test_coroutine_e2e.py create mode 100644 tests/azure-functions-durable/test_async_compat_context.py create mode 100644 tests/azure-functions-durable/test_functions_coroutine_orchestrators.py diff --git a/azure-functions-durable/azure/durable_functions/__init__.py b/azure-functions-durable/azure/durable_functions/__init__.py index de861727..1e4a0fa7 100644 --- a/azure-functions-durable/azure/durable_functions/__init__.py +++ b/azure-functions-durable/azure/durable_functions/__init__.py @@ -17,7 +17,10 @@ from .internal.compat.entity_state_response import EntityStateResponse from .internal.compat.entity_id import EntityId from .internal.compat.token_source import ManagedIdentityTokenSource, TokenSource -from .internal.compat.orchestration_context import DurableOrchestrationContext +from .internal.compat.orchestration_context import ( + AsyncDurableOrchestrationContext, + DurableOrchestrationContext, +) from .internal.compat.entity_context import DurableEntityContext from .internal.compat.compat_aliases import ( DurableOrchestrationClient, @@ -39,6 +42,7 @@ __all__ = [ "Blueprint", + "AsyncDurableOrchestrationContext", "DFApp", "DurableEntityContext", "DurableFunctionsClient", diff --git a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py index 4d7df348..be5e8d1d 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -21,6 +21,7 @@ builtin_http_poll_orchestrator, ) from ..internal.compat.activity import wrap_activity, wrap_activity_payloads +from ..internal.compat.orchestration_context import FunctionsOrchestrator from ..internal.invocation import preserve_function_source, wrap_invocation from ..worker import DurableFunctionsWorker @@ -169,7 +170,7 @@ def _configure_orchestrator_callable( wrap: Callable[[Callable[..., Any]], FunctionBuilder], context_name: str, input_type: Optional[type] = None - ) -> Callable[[task.Orchestrator[Any, Any]], FunctionBuilder]: + ) -> Callable[[FunctionsOrchestrator], FunctionBuilder]: """Obtain decorator to construct an Orchestrator class from a user-defined Function. Parameters @@ -187,7 +188,7 @@ def _configure_orchestrator_callable( The function to construct an Orchestrator class from the user-defined Function, wrapped by the next decorator in the sequence. """ - def decorator(orchestrator_func: task.Orchestrator[Any, Any]) -> FunctionBuilder: + def decorator(orchestrator_func: FunctionsOrchestrator) -> FunctionBuilder: # Construct an orchestrator based on the end-user code if input_type is not None: @@ -281,7 +282,7 @@ def decorator(func: Callable[..., Any]) -> FunctionBuilder: def orchestration_trigger(self, context_name: str, orchestration: Optional[str] = None, input_type: Optional[type] = None - ) -> Callable[[task.Orchestrator[Any, Any]], FunctionBuilder]: + ) -> Callable[[FunctionsOrchestrator], FunctionBuilder]: """Register an Orchestrator Function. Parameters diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py index 84561e39..ce05f38a 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py @@ -1,22 +1,25 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. +import asyncio import inspect import json -from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, cast +from collections.abc import Awaitable, Callable, Coroutine, Generator, Sequence +from datetime import UTC, datetime, timedelta +from typing import Any, Optional, TypeAlias, TypeVar, cast, overload from uuid import UUID from durabletask import task from durabletask.entities import EntityInstanceId +from durabletask.internal.shared import is_async_callable from durabletask.task import OrchestrationContext, RetryPolicy, Task +from ...http.models import DurableHttpResponse, HttpRetryOptions from ..serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER from .function_context import FunctionContext from .token_source import TokenSource -if TYPE_CHECKING: - from ...http.models import HttpRetryOptions +T = TypeVar("T") class DurableOrchestrationContext: @@ -277,6 +280,271 @@ def call_http(self, return_type=DurableHttpResponse) +def _schedule_awaitables( + operations: Sequence[Awaitable[T]], +) -> tuple[list[asyncio.Future[T]], list[asyncio.Future[T]]]: + """Schedule each distinct awaitable once, retaining input order.""" + scheduled: dict[int, asyncio.Future[T]] = {} + owned: list[asyncio.Future[T]] = [] + futures: list[asyncio.Future[T]] = [] + loop = asyncio.get_running_loop() + for operation in operations: + if not inspect.isawaitable(operation): + raise TypeError("Orchestration operations must be awaitable.") + if isinstance(operation, asyncio.Future) and operation.get_loop() is not loop: + raise ValueError("Orchestration tasks must belong to the current durable event loop.") + for operation in operations: + key = id(operation) + if key not in scheduled: + future = asyncio.ensure_future(operation) + scheduled[key] = future + if future is not operation: + owned.append(future) + futures.append(scheduled[key]) + return futures, owned + + +async def _cancel_owned(futures: Sequence[asyncio.Future[Any]]) -> None: + pending = [future for future in futures if not future.done()] + for future in pending: + future.cancel() + if pending: + await asyncio.gather(*pending, return_exceptions=True) + + +class AsyncDurableOrchestrationContext: + """Functions v1-style context supplied automatically to async orchestrators. + + Durable operations are cold, single-use coroutines. Synchronous commands + and metadata have the same behavior and argument names as the generator + context. Use :meth:`as_task_context` only when eager durable task handles + are needed for interoperability. + """ + + def __init__( + self, ctx: task.AsyncOrchestrationContext, + orchestration_input: Any = None, input_type: type | None = None, + ) -> None: + self._ctx = ctx + self._compat = DurableOrchestrationContext(ctx.as_task_context(), orchestration_input, input_type) + + @overload + def get_input(self, expected_type: type[T]) -> T: + ... + + @overload + def get_input(self, expected_type: None = None) -> Any: + ... + + def get_input(self, expected_type: type | None = None) -> Any: + """Get the orchestration input, optionally coerced to the expected type.""" + return self._compat.get_input(expected_type) + + @property + def instance_id(self) -> str: + """Get the current orchestration instance ID.""" + return self._compat.instance_id + + @property + def parent_instance_id(self) -> str | None: + """Get the parent orchestration instance ID, if any.""" + return self._compat.parent_instance_id + + @property + def version(self) -> str | None: + """Get the orchestration instance version, if any.""" + return self._compat.version + + @property + def current_utc_datetime(self) -> datetime: + """Get the replay-safe current UTC time as a timezone-aware datetime.""" + return self._compat.current_utc_datetime + + @property + def is_replaying(self) -> bool: + """Get whether the orchestration is replaying its history.""" + return self._compat.is_replaying + + @property + def custom_status(self) -> Any: + """Get the custom status set during this execution.""" + return self._compat.custom_status + + @property + def will_continue_as_new(self) -> bool: + """Get whether continue-as-new was requested during this execution.""" + return self._compat.will_continue_as_new + + @property + def function_context(self) -> FunctionContext: + """Get the Functions context bag.""" + return self._compat.function_context + + @property + def histories(self) -> Any: + """History is not exposed on orchestration contexts.""" + return self._compat.histories + + def continue_as_new(self, input_: Any) -> None: + """Restart the orchestration with a new input.""" + self._compat.continue_as_new(input_) + + def set_custom_status(self, status: Any) -> None: + """Set the orchestration's custom status.""" + self._compat.set_custom_status(status) + + def new_uuid(self) -> str: + """Create a replay-safe UUID string.""" + return self._compat.new_uuid() + + def new_guid(self) -> UUID: + """Create a replay-safe UUID.""" + return self._compat.new_guid() + + def signal_entity( + self, entityId: EntityInstanceId, operationName: str, operationInput: Any = None, + ) -> None: + """Signal an entity without waiting for a result.""" + self._compat.signal_entity(entityId, operationName, operationInput) + + def as_task_context(self) -> DurableOrchestrationContext: + """Get the v1-compatible eager task-handle context.""" + return self._compat + + async def call_activity(self, name: Callable[..., Any] | str, input_: Any = None) -> Any: + """Start and await an activity.""" + return await self._ctx.call_activity(name, input=input_) + + async def call_activity_with_retry( + self, name: Callable[..., Any] | str, retry_options: RetryPolicy, input_: Any = None, + ) -> Any: + """Start and await an activity with retries.""" + return await self._ctx.call_activity(name, input=input_, retry_policy=retry_options) + + async def call_sub_orchestrator( + self, name: Callable[..., Any] | str, input_: Any = None, instance_id: str | None = None, + ) -> Any: + """Start and await a sub-orchestration.""" + return await self._ctx.call_sub_orchestrator(name, input=input_, instance_id=instance_id) + + async def call_sub_orchestrator_with_retry( + self, name: Callable[..., Any] | str, retry_options: RetryPolicy, + input_: Any = None, instance_id: str | None = None, + ) -> Any: + """Start and await a sub-orchestration with retries.""" + return await self._ctx.call_sub_orchestrator( + name, input=input_, instance_id=instance_id, retry_policy=retry_options) + + async def create_timer(self, fire_at: datetime | timedelta) -> None: + """Wait for a durable timer, withdrawing it locally if cancelled.""" + await self._ctx.create_timer(fire_at) + + @overload + async def wait_for_external_event( + self, name: str, expected_type: type[T], *, timeout: timedelta | None = None, + ) -> T: + ... + + @overload + async def wait_for_external_event( + self, name: str, expected_type: None = None, *, timeout: timedelta | None = None, + ) -> Any: + ... + + async def wait_for_external_event( + self, name: str, expected_type: type[Any] | None = None, *, timeout: timedelta | None = None, + ) -> Any: + """Wait for an event, withdrawing the local subscription on cancellation or timeout.""" + return await self._ctx.wait_for_external_event(name, data_type=expected_type, timeout=timeout) + + async def call_entity( + self, entityId: EntityInstanceId, operationName: str, operationInput: Any = None, + ) -> Any: + """Start and await an entity operation.""" + return await self._ctx.call_entity(entityId, operationName, operationInput) + + async def call_http( + self, method: str, uri: str, content: Any = None, headers: dict[str, str] | None = None, + token_source: TokenSource | None = None, is_raw_str: bool = False, + retry_options: HttpRetryOptions | None = None, + ) -> DurableHttpResponse: + """Start and await a durable HTTP request.""" + return await self._compat.call_http( + method, uri, content, headers, token_source, is_raw_str, retry_options) + + async def task_all(self, operations: Sequence[Awaitable[T]]) -> list[T]: + """Wait for every operation, preserving input order and duplicate references. + + The first observed failure is raised after all operations finish. + Existing asyncio futures are borrowed; cancelling this group leaves + them running. Coroutines started by this group are cancelled and + cleaned up if the group itself is cancelled. + """ + futures, owned = _schedule_awaitables(operations) + unique = list({id(future): future for future in futures}.values()) + failures: list[BaseException] = [] + + def record_failure(future: asyncio.Future[T]) -> None: + if not failures: + if future.cancelled(): + failures.append(asyncio.CancelledError()) + else: + error = future.exception() + if error is not None: + failures.append(error) + + for future in unique: + future.add_done_callback(record_failure) + try: + results = await asyncio.gather(*(asyncio.shield(future) for future in futures), + return_exceptions=True) + if failures: + raise failures[0] + return cast(list[T], results) + except asyncio.CancelledError: + await _cancel_owned(owned) + raise + finally: + for future in unique: + future.remove_done_callback(record_failure) + + async def task_any(self, operations: Sequence[Awaitable[T]]) -> asyncio.Future[T]: + """Return the first completed operation without cancelling the losers. + + The winner is the original asyncio future if one was supplied; a raw + coroutine is wrapped in a new task. Use tasks when identity or later + access to losing operations is needed. Cancelling the group cancels + only coroutines it started, not futures supplied by its caller. + """ + if not operations: + raise ValueError("task_any requires at least one operation.") + futures, owned = _schedule_awaitables(operations) + unique = list({id(future): future for future in futures}.values()) + winner: asyncio.Future[asyncio.Future[T]] = asyncio.get_running_loop().create_future() + + def completed(future: asyncio.Future[T]) -> None: + if not winner.done(): + winner.set_result(future) + + for future in unique: + future.add_done_callback(completed) + try: + return await winner + except asyncio.CancelledError: + await _cancel_owned(owned) + raise + finally: + for future in unique: + future.remove_done_callback(completed) + + +FunctionsOrchestrator: TypeAlias = ( + task.Orchestrator[Any, Any] + | Callable[[DurableOrchestrationContext], Any] + | Callable[[AsyncDurableOrchestrationContext], Coroutine[Any, Any, Any]] +) + + def accepts_two_positional_args(fn: Callable[..., Any]) -> bool: """Return True if ``fn`` can be called with two positional args ``(ctx, input)``. @@ -326,6 +594,20 @@ def wrap_orchestrator(fn: Callable[..., Any]) -> Callable[..., Any]: input_type = getattr(fn, "_df_input_type", None) name = getattr(fn, "__name__", "orchestrator") + if is_async_callable(fn): + async def _async_wrapper(context: task.AsyncOrchestrationContext, _input: Any = None) -> Any: + adapter = AsyncDurableOrchestrationContext(context, _input, input_type) + coroutine = fn(adapter) + if not inspect.iscoroutine(coroutine): + if inspect.isgenerator(coroutine): + coroutine.close() + raise TypeError("An async orchestrator must return a coroutine.") + return await coroutine + _async_wrapper.__name__ = name + if input_type is not None: + _async_wrapper.__annotations__["_input"] = input_type + return _async_wrapper + if inspect.isgeneratorfunction(fn): def _generator_wrapper(context: OrchestrationContext, _input: Any = None) -> Any: adapter = DurableOrchestrationContext(context, _input, input_type) diff --git a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py index 43daf0c5..491b10c0 100644 --- a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py +++ b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py @@ -1,16 +1,18 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Replay simulation helper for unit-testing orchestrator generators.""" +"""Replay simulation helper for unit-testing generator or coroutine orchestrators.""" -from typing import Any, Generator +import asyncio +from collections.abc import Coroutine, Generator +from typing import Any def orchestrator_generator_wrapper( - generator: Generator[Any, Any, Any]) -> Generator[Any, None, None]: - """Drive a user orchestrator generator the way the Durable replay engine does. + generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]) -> Generator[Any, None, None]: + """Drive a user orchestrator the way the Durable replay engine does. - A Durable Functions orchestrator is a generator that yields task objects and + A Durable Functions orchestrator yields or awaits durable task objects and is resumed with each task's result until it returns the orchestration output. In a real execution the Durable Task runtime performs this drive loop during replay. This wrapper reproduces that loop in-process so an @@ -20,10 +22,14 @@ def orchestrator_generator_wrapper( back into the orchestrator (mirroring how failed activities, timers, or other tasks surface during replay). + This mock-based helper handles direct durable awaits in synchronous or + async tests. It does not simulate the runtime's asyncio scheduler; use + replay/backend tests for gather, task groups, sleeps and timeouts. + Parameters ---------- - generator: Generator[Any, Any, Any] - The user orchestrator generator. It is expected to yield task objects + generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any] + The user orchestrator generator or coroutine. It must suspend on tasks that expose a ``result`` attribute and to be resumed with those results until it returns the orchestration output. @@ -33,14 +39,24 @@ def orchestrator_generator_wrapper( Each task object yielded by the orchestrator, followed by the orchestrator's final return value. """ + def step(value: Any = None, *, error: Exception | None = None) -> Any: + previous_loop = asyncio._get_running_loop() + try: + if isinstance(generator, Coroutine): + # Direct durable awaits belong to this mock driver, not the + # caller's asyncio loop (or an enclosing durable dispatch). + asyncio._set_running_loop(None) + return generator.throw(error) if error is not None else generator.send(value) + finally: + asyncio._set_running_loop(previous_loop) + try: - previous = next(generator) + previous = step() except StopIteration as e: # An orchestrator that returns immediately without yielding any task # completes right away (mirroring the runtime, which finishes the # orchestration on the first return). Surface its output as the final - # value and stop -- ``next`` raising here otherwise leaks as an - # unhandled ``StopIteration``. + # value and stop -- letting StopIteration escape would become RuntimeError. yield e.value return yield previous @@ -52,9 +68,9 @@ def orchestrator_generator_wrapper( # Simulated activity exceptions, timer-interrupted exceptions, # or any other case where reading a task result would throw: # push the exception back into the orchestrator as replay does. - previous = generator.throw(e) + previous = step(error=e) else: - previous = generator.send(previous_result) + previous = step(previous_result) yield previous except StopIteration as e: # The orchestrator returned; surface its output as the final value. diff --git a/azure-functions-durable/azure/durable_functions/worker.py b/azure-functions-durable/azure/durable_functions/worker.py index 3a569d93..43db9156 100644 --- a/azure-functions-durable/azure/durable_functions/worker.py +++ b/azure-functions-durable/azure/durable_functions/worker.py @@ -22,7 +22,7 @@ from .internal.azurefunctions_null_stub import AzureFunctionsNullStub from .internal.invocation import run_sync from .internal.compat.entity_context import wrap_entity -from .internal.compat.orchestration_context import wrap_orchestrator +from .internal.compat.orchestration_context import FunctionsOrchestrator, wrap_orchestrator from .internal.payloads import ( discard_scheduled_activity_inputs, get_transport_payload_store, hydrate_entity_request, hydrate_entity_request_async, @@ -60,7 +60,7 @@ def __init__(self) -> None: ) self._registration_lock = Lock() self._registered_orchestrator_functions: dict[ - str, task.Orchestrator[Any, Any] + str, FunctionsOrchestrator ] = {} self._registered_entity_functions: dict[ str, task.Entity[Any, Any] @@ -72,7 +72,7 @@ def add_named_orchestrator(self, name: str, func: task.Orchestrator[Any, Any]) - def _register_orchestrator_once( self, name: str, - func: task.Orchestrator[Any, Any], + func: FunctionsOrchestrator, ) -> None: with self._registration_lock: if self._registered_orchestrator_functions.get(name) is func: @@ -92,7 +92,7 @@ def _register_entity_once(self, func: task.Entity[Any, Any]) -> None: self.add_entity(wrap_entity(func), name=name) self._registered_entity_functions[name] = func - def execute_orchestration_request(self, func: task.Orchestrator[Any, Any], context: Any) -> str: + def execute_orchestration_request(self, func: FunctionsOrchestrator, context: Any) -> str: payload_store = get_transport_payload_store() context_body = getattr(context, "body", None) if context_body is None: @@ -109,7 +109,7 @@ def execute_orchestration_request(self, func: task.Orchestrator[Any, Any], conte externalize_payloads(response, payload_store, instance_id=request.instanceId) return base64.b64encode(response.SerializeToString()).decode("utf-8") - async def execute_orchestration_request_async(self, func: task.Orchestrator[Any, Any], context: Any) -> str: + async def execute_orchestration_request_async(self, func: FunctionsOrchestrator, context: Any) -> str: payload_store = get_transport_payload_store() request = OrchestratorRequest() request.ParseFromString(base64.b64decode(getattr(context, "body", None) or context)) @@ -123,7 +123,7 @@ async def execute_orchestration_request_async(self, func: task.Orchestrator[Any, return base64.b64encode(response.SerializeToString()).decode("utf-8") def _run_orchestration( - self, func: task.Orchestrator[Any, Any], request: OrchestratorRequest) -> OrchestratorResponse: + self, func: FunctionsOrchestrator, request: OrchestratorRequest) -> OrchestratorResponse: stub: Any = AzureFunctionsNullStub() response: Optional[OrchestratorResponse] = None diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 34b44d98..6a6c67ae 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -13,6 +13,7 @@ """ from datetime import timedelta +import asyncio import logging from typing import Any @@ -35,6 +36,27 @@ def activity_chain(ctx: task.OrchestrationContext, _: Any): return [first, second, third] +@bp.orchestration_trigger(context_name="context") +async def async_await_probe(ctx: task.AsyncOrchestrationContext, _: Any) -> dict[str, Any]: + greeting = await ctx.call_activity("say_hello", input="Tokyo", return_type=str) + values = await asyncio.gather(*(ctx.call_activity("square", input=i, return_type=int) for i in (2, 3))) + await asyncio.sleep(0) + await asyncio.sleep(0.01) + child: list[str] = await ctx.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": sum(values), "child": child} + + +@bp.orchestration_trigger(context_name="context") +async def async_cold_probe(ctx: task.AsyncOrchestrationContext, _: Any) -> dict[str, Any]: + greeting = await asyncio.create_task(ctx.call_activity("say_hello", input="Tokyo", return_type=str)) + async with asyncio.TaskGroup() as group: + first = group.create_task(ctx.call_activity("square", input=2, return_type=int)) + second = group.create_task(ctx.call_activity("square", input=3, return_type=int)) + await ctx.create_timer(timedelta(milliseconds=10)) + child: list[str] = await ctx.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": first.result() + second.result(), "child": child} + + @bp.orchestration_trigger(context_name="context") def logging_filtered(ctx: task.OrchestrationContext, _: Any): _LOGGER.info("worker-info-filter-anchor %s", ctx.instance_id) diff --git a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py index a96467c7..1477caae 100644 --- a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py @@ -13,6 +13,8 @@ """ from datetime import timedelta +import asyncio +from typing import Any import azure.durable_functions as df @@ -31,6 +33,27 @@ def activity_chain(context: df.DurableOrchestrationContext): return [first, second, third] +@bp.orchestration_trigger(context_name="context") +async def async_await_probe(context: df.AsyncDurableOrchestrationContext) -> dict[str, Any]: + greeting = await context.call_activity("say_hello", "Tokyo") + values = await asyncio.gather(*(context.call_activity("square", i) for i in (2, 3))) + await asyncio.sleep(0) + await asyncio.sleep(0.01) + child = await context.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": sum(values), "child": child} + + +@bp.orchestration_trigger(context_name="context") +async def async_cold_probe(context: df.AsyncDurableOrchestrationContext) -> dict[str, Any]: + greeting = await asyncio.create_task(context.call_activity("say_hello", "Tokyo")) + async with asyncio.TaskGroup() as group: + first = group.create_task(context.call_activity("square", 2)) + second = group.create_task(context.call_activity("square", 3)) + await context.create_timer(timedelta(milliseconds=10)) + child: list[str] = await context.call_sub_orchestrator("activity_chain") + return {"greeting": greeting, "total": first.result() + second.result(), "child": child} + + @bp.orchestration_trigger(context_name="context") def fan_out_fan_in(context: df.DurableOrchestrationContext): count = context.get_input() or 5 diff --git a/tests/azure-functions-durable/e2e/test_coroutine_e2e.py b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py new file mode 100644 index 00000000..6e5f1629 --- /dev/null +++ b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py @@ -0,0 +1,25 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Real-host smoke coverage for both async orchestrator authoring styles.""" + +import pytest + +from ._harness import FunctionApp + + +pytestmark = pytest.mark.functions_e2e + + +@pytest.mark.parametrize("app_fixture", ["dtask_app", "v1_app"]) +@pytest.mark.parametrize("orchestrator", ["async_await_probe", "async_cold_probe"]) +def test_async_await_orchestrator(request: pytest.FixtureRequest, app_fixture: str, orchestrator: str) -> None: + app: FunctionApp = request.getfixturevalue(app_fixture) + instance_id = app.start_orchestration(orchestrator) + status = app.wait_for_completion(instance_id) + assert status["runtimeStatus"].upper() == "COMPLETED", status + assert status["output"] == { + "greeting": "Hello Tokyo!", + "total": 13, + "child": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"], + } diff --git a/tests/azure-functions-durable/test_async_compat_context.py b/tests/azure-functions-durable/test_async_compat_context.py new file mode 100644 index 00000000..27b7b6fe --- /dev/null +++ b/tests/azure-functions-durable/test_async_compat_context.py @@ -0,0 +1,656 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Replay and ownership tests for the Functions coroutine compatibility context.""" + +import asyncio +import base64 +import inspect +import json +from collections.abc import Callable, Coroutine, Generator +from datetime import UTC, datetime, timedelta +from functools import partial, wraps +from typing import Any +from uuid import UUID + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER +from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import entities, task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +INSTANCE_ID = "async-compat-test" +NOW = datetime(2026, 1, 1) + + +class TypedInput: + def __init__(self, value: int) -> None: + self.value = value + + def to_json(self) -> dict[str, int]: + return {"value": self.value} + + @classmethod + def from_json(cls, data: dict[str, int]) -> "TypedInput": + return cls(data["value"]) + + +def _start() -> list[pb.HistoryEvent]: + return [ + helpers.new_orchestrator_started_event(NOW), + helpers.new_execution_started_event("compat_test", INSTANCE_ID, "null"), + ] + + +def _execute( + orchestrator: Callable[..., Any], + old: list[pb.HistoryEvent], + new: list[pb.HistoryEvent], +) -> pb.OrchestratorResponse: + request = pb.OrchestratorRequest(instanceId=INSTANCE_ID, pastEvents=old, newEvents=new) + encoded = base64.b64encode(request.SerializeToString()).decode() + + def invoke() -> pb.OrchestratorResponse: + response = pb.OrchestratorResponse() + result = DurableFunctionsWorker().execute_orchestration_request(orchestrator, encoded) + response.ParseFromString(base64.b64decode(result)) + return response + + response = invoke() + assert invoke().actions == response.actions + return response + + +def _completed(response: pb.OrchestratorResponse) -> Any: + assert len(response.actions) == 1 + assert response.actions[0].HasField("completeOrchestration") + completed = response.actions[0].completeOrchestration + assert completed.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, completed.failureDetails + return json.loads(completed.result.value) + + +def test_task_all_preserves_duplicates_order_and_shared_tasks() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + shared = asyncio.create_task(ctx.call_activity("shared")) + repeated = ctx.call_activity("repeated") + left = asyncio.create_task(ctx.task_all([repeated, shared, repeated])) + right = asyncio.create_task(ctx.task_all([shared])) + return list(await asyncio.gather(left, right)) + + start = _start() + initial = _execute(orchestrator, [], start) + assert [(action.id, action.scheduleTask.name) for action in initial.actions] == [ + (1, "shared"), (2, "repeated")] + history = start + [ + helpers.new_task_scheduled_event(1, "shared"), + helpers.new_task_scheduled_event(2, "repeated"), + ] + repeated_done = [helpers.new_task_completed_event(2, '"R"')] + assert not _execute(orchestrator, history, repeated_done).actions + shared_done = [helpers.new_task_completed_event(1, '"S"')] + assert _completed(_execute(orchestrator, history + repeated_done, shared_done)) == [ + ["R", "S", "R"], ["S"]] + + +def test_cancelling_one_group_does_not_cancel_another_groups_shared_task() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + shared = asyncio.create_task(ctx.wait_for_external_event("shared")) + cancelled = asyncio.create_task(ctx.task_all([shared])) + survivor = asyncio.create_task(ctx.task_all([shared, shared])) + await asyncio.sleep(0) + await asyncio.sleep(0) + cancelled.cancel() + try: + await cancelled + except asyncio.CancelledError: + pass + else: + raise AssertionError("Expected group cancellation") + return await survivor + + start = _start() + assert not _execute(orchestrator, [], start).actions + assert _completed(_execute( + orchestrator, start, [helpers.new_event_raised_event("shared", '"value"')])) == ["value", "value"] + + +def test_task_all_waits_for_every_child_and_raises_first_observed_failure() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + try: + await ctx.task_all([ + ctx.call_activity("first"), + ctx.call_activity("second"), + ctx.call_activity("last"), + ]) + except task.TaskFailedError as error: + return str(error) + raise AssertionError("Expected the group to fail") + + start = _start() + assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == [ + "first", "second", "last"] + history = start + [ + helpers.new_task_scheduled_event(1, "first"), + helpers.new_task_scheduled_event(2, "second"), + helpers.new_task_scheduled_event(3, "last"), + ] + second_failed = [helpers.new_task_failed_event(2, ValueError("second failure"))] + first_failed = [helpers.new_task_failed_event(1, ValueError("first failure"))] + assert not _execute(orchestrator, history, second_failed).actions + assert not _execute(orchestrator, history + second_failed, first_failed).actions + result = _completed(_execute( + orchestrator, history + second_failed + first_failed, + [helpers.new_task_completed_event(3, '"done"')])) + assert "second failure" in result + assert "first failure" not in result + + +@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) +def test_group_failure_order_follows_completions_in_one_history_batch(group_method: str) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + first = asyncio.create_task(ctx.call_activity("first")) + second = asyncio.create_task(ctx.call_activity("second")) + try: + if group_method == "task_all": + await ctx.task_all([first, second]) + else: + winner = await ctx.task_any([first, second]) + assert winner is second + try: + await winner + finally: + try: + await first + except task.TaskFailedError: + pass + except task.TaskFailedError as error: + return str(error) + raise AssertionError("Expected the group to fail") + + start = _start() + assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["first", "second"] + history = start + [ + helpers.new_task_scheduled_event(1, "first"), + helpers.new_task_scheduled_event(2, "second"), + ] + failures = [ + helpers.new_task_failed_event(2, ValueError("second failure")), + helpers.new_task_failed_event(1, ValueError("first failure")), + ] + result = _completed(_execute(orchestrator, history, failures)) + assert "second failure" in result + assert "first failure" not in result + + +@pytest.mark.parametrize("fail", [False, True]) +def test_task_any_returns_original_winner_and_keeps_loser_running(fail: bool) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + slow = asyncio.create_task(ctx.call_activity("slow")) + fast = asyncio.create_task(ctx.call_activity("fast")) + winner = await ctx.task_any([slow, fast, slow]) + assert winner is fast + assert not slow.cancelled() + try: + value = await winner + except task.TaskFailedError: + value = "failed" + return [value, await slow] + + start = _start() + assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["slow", "fast"] + history = start + [ + helpers.new_task_scheduled_event(1, "slow"), + helpers.new_task_scheduled_event(2, "fast"), + ] + fast_done = [helpers.new_task_failed_event(2, ValueError("fast failure"))] if fail else [ + helpers.new_task_completed_event(2, '"fast"')] + assert not _execute(orchestrator, history, fast_done).actions + slow_done = [helpers.new_task_completed_event(1, '"slow"')] + assert _completed(_execute(orchestrator, history + fast_done, slow_done)) == [ + "failed" if fail else "fast", "slow"] + + +def test_task_any_keeps_owned_coroutine_loser_running() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + async def slower() -> None: + await ctx.call_activity("slow") + await ctx.call_activity("continued") + + winner = await ctx.task_any([slower(), ctx.call_activity("fast")]) + assert await winner == "fast" + return await ctx.wait_for_external_event("finish", str) + + start = _start() + assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["slow", "fast"] + history = start + [ + helpers.new_task_scheduled_event(1, "slow"), + helpers.new_task_scheduled_event(2, "fast"), + ] + fast_done = [helpers.new_task_completed_event(2, '"fast"')] + assert not _execute(orchestrator, history, fast_done).actions + slow_done = [helpers.new_task_completed_event(1, '"slow"')] + continued = _execute(orchestrator, history + fast_done, slow_done) + assert [(a.id, a.scheduleTask.name) for a in continued.actions] == [(3, "continued")] + history += fast_done + slow_done + [ + helpers.new_task_scheduled_event(3, "continued"), + helpers.new_task_completed_event(3, '"done"'), + ] + assert _completed(_execute( + orchestrator, history, [helpers.new_event_raised_event("finish", '"done"')])) == "done" + + +@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) +def test_cancelling_group_withdraws_owned_events_but_not_borrowed_tasks(group_method: str) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + borrowed = asyncio.create_task(ctx.wait_for_external_event("borrowed")) + operations = [ctx.wait_for_external_event("owned"), borrowed] + group = asyncio.create_task(getattr(ctx, group_method)(operations)) + await asyncio.sleep(0) + await asyncio.sleep(0) + group.cancel() + try: + await group + except asyncio.CancelledError: + pass + else: + raise AssertionError("Expected group cancellation") + assert not borrowed.done() + replacement = await ctx.wait_for_external_event("owned") + return [replacement, await borrowed] + + start = _start() + assert not _execute(orchestrator, [], start).actions + events = [ + helpers.new_event_raised_event("owned", '"replacement"'), + helpers.new_event_raised_event("borrowed", '"borrowed"'), + ] + assert _completed(_execute(orchestrator, start, events)) == ["replacement", "borrowed"] + + +@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) +def test_cancelling_group_withdraws_owned_timers_but_not_borrowed_tasks(group_method: str) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + borrowed = asyncio.create_task(ctx.create_timer(timedelta(seconds=5))) + operations = [ctx.create_timer(timedelta(seconds=10)), borrowed] + group = asyncio.create_task(getattr(ctx, group_method)(operations)) + await asyncio.sleep(0) + await asyncio.sleep(0) + group.cancel() + try: + await group + except asyncio.CancelledError: + pass + else: + raise AssertionError("Expected group cancellation") + assert not borrowed.cancelled() + await borrowed + return "survived" + + start = _start() + initial = _execute(orchestrator, [], start) + assert len(initial.actions) == 1 + timer = initial.actions[0] + assert timer.createTimer.fireAt.ToDatetime() == NOW + timedelta(seconds=5) + deadline = NOW + timedelta(seconds=5) + history = start + [helpers.new_timer_created_event(timer.id, deadline)] + fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer.id, deadline)] + assert _completed(_execute(orchestrator, history, fired)) == "survived" + + +@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) +def test_group_timeout_withdraws_owned_event_wait(group_method: str) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + try: + await asyncio.wait_for( + getattr(ctx, group_method)([ctx.wait_for_external_event("approval")]), timeout=1) + except TimeoutError: + return await ctx.wait_for_external_event("approval", str) + raise AssertionError("Expected timeout") + + start = _start() + initial = _execute(orchestrator, [], start) + assert len(initial.actions) == 1 + timer = initial.actions[0] + deadline = NOW + timedelta(seconds=1) + assert timer.createTimer.fireAt.ToDatetime() == deadline + history = start + [helpers.new_timer_created_event(timer.id, deadline)] + timeout = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer.id, deadline)] + assert not _execute(orchestrator, history, timeout).actions + assert _completed(_execute( + orchestrator, history + timeout, [helpers.new_event_raised_event("approval", '"approved"')])) == "approved" + + +def test_empty_and_invalid_groups() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + assert await ctx.task_all([]) == [] + errors: list[str] = [] + invalid: Any = object() + operations: list[Coroutine[Any, Any, Any]] = [ + ctx.task_any([]), ctx.task_all([invalid]), ctx.task_any([invalid])] + for operation in operations: + try: + await operation + except (TypeError, ValueError) as error: + errors.append(str(error)) + return errors + + result = _completed(_execute(orchestrator, [], _start())) + assert result == [ + "task_any requires at least one operation.", + "Orchestration operations must be awaitable.", + "Orchestration operations must be awaitable.", + ] + + +@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) +def test_invalid_group_does_not_start_its_other_operations(group_method: str) -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + operation = ctx.call_activity("work") + try: + await getattr(ctx, group_method)([operation, object()]) + except TypeError as error: + assert str(error) == "Orchestration operations must be awaitable." + else: + raise AssertionError("Expected invalid group to fail") + return await operation + + start = _start() + initial = _execute(orchestrator, [], start) + assert [(a.id, a.scheduleTask.name) for a in initial.actions] == [(1, "work")] + history = start + [helpers.new_task_scheduled_event(1, "work")] + assert _completed(_execute( + orchestrator, history, [helpers.new_task_completed_event(1, '"done"')])) == "done" + + +@pytest.mark.parametrize("shape", ["function", "wrapped", "partial", "object", "partial-object"]) +def test_v1_async_callable_shapes_receive_compatibility_context(shape: str) -> None: + async def plain(ctx: df.AsyncDurableOrchestrationContext) -> Any: + assert isinstance(ctx, df.AsyncDurableOrchestrationContext) + return await ctx.call_activity("work", ctx.get_input()) + + @wraps(plain) + def wrapped(*args: Any, **kwargs: Any) -> Any: + return plain(*args, **kwargs) + + async def with_prefix(prefix: str, ctx: df.AsyncDurableOrchestrationContext) -> Any: + assert isinstance(ctx, df.AsyncDurableOrchestrationContext) + return await ctx.call_activity("work", prefix) + + class AsyncCallable: + async def __call__(self, ctx: df.AsyncDurableOrchestrationContext) -> Any: + return await plain(ctx) + + class PartialCallable: + async def __call__(self, prefix: str, ctx: df.AsyncDurableOrchestrationContext) -> Any: + return await with_prefix(prefix, ctx) + + variants = { + "function": plain, + "wrapped": wrapped, + "partial": partial(with_prefix, "value"), + "object": AsyncCallable(), + "partial-object": partial(PartialCallable(), "value"), + } + fn = variants[shape] + start = _start() + initial = _execute(fn, [], start) + assert len(initial.actions) == 1 + assert initial.actions[0].scheduleTask.name == "work" + history = start + [helpers.new_task_scheduled_event(1, "work")] + assert _completed(_execute(fn, history, [helpers.new_task_completed_event(1, '"done"')])) == "done" + + +def test_unmarked_v1_wrapper_returning_coroutine_fails_and_closes_it() -> None: + created: list[Any] = [] + + async def async_orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: + return "unexpected" + + def wrapper(ctx: df.DurableOrchestrationContext) -> Any: + coroutine = async_orchestrator(ctx) # type: ignore[arg-type] + created.append(coroutine) + return coroutine + + result = _execute(wrapper, [], _start()) + failure = result.actions[0].completeOrchestration + assert failure.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "synchronous orchestrator returned a coroutine" in failure.failureDetails.errorMessage + assert all(inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED for coroutine in created) + + +def test_metadata_status_and_eager_context_share_state() -> None: + def inspect_context(ctx: df.DurableOrchestrationContext | df.AsyncDurableOrchestrationContext) -> list[Any]: + ctx.set_custom_status({"step": 1}) + assert ctx.custom_status == {"step": 1} + assert ctx.get_input(dict) == {"number": 7} + assert vars(ctx.function_context) == {} + with pytest.raises(NotImplementedError): + _ = ctx.histories + uuid = ctx.new_uuid() + guid = ctx.new_guid() + assert isinstance(guid, UUID) + return [ + ctx.instance_id, + ctx.parent_instance_id, + ctx.version, + ctx.is_replaying, + ctx.current_utc_datetime.isoformat(), + uuid, + str(guid), + ] + + async def coroutine(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + result = inspect_context(ctx) + eager = ctx.as_task_context() + assert eager.custom_status == ctx.custom_status + eager.set_custom_status({"step": 2}) + assert ctx.custom_status == {"step": 2} + return result + + def generator_style(ctx: df.DurableOrchestrationContext) -> list[Any]: + result = inspect_context(ctx) + ctx.set_custom_status({"step": 2}) + return result + + started = helpers.new_execution_started_event( + "compat_test", INSTANCE_ID, '{"number": 7}', version="v1") + started.executionStarted.parentInstance.CopyFrom(pb.ParentInstanceInfo( + taskScheduledId=1, + orchestrationInstance=pb.OrchestrationInstance(instanceId="parent"))) + start = [helpers.new_orchestrator_started_event(NOW), started] + result = _execute(coroutine, [], start) + assert result.actions == _execute(generator_style, [], start).actions + values = _completed(result) + assert values[:5] == [INSTANCE_ID, "parent", "v1", False, NOW.replace(tzinfo=UTC).isoformat()] + assert str(UUID(values[5])) == values[5] + assert str(UUID(values[6])) == values[6] + assert json.loads(result.customStatus.value) == {"step": 2} + + +@pytest.mark.parametrize("strict", [False, True]) +async def test_decorator_input_type_reaches_async_compat_context(monkeypatch: pytest.MonkeyPatch, strict: bool) -> None: + monkeypatch.setenv("AZURE_FUNCTIONS_DURABLE_STRICT_TYPING", "1" if strict else "0") + app = df.DFApp() + + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> int: + typed = ctx.get_input() + assert isinstance(typed, TypedInput) + assert isinstance(ctx.get_input(TypedInput), TypedInput) + return typed.value + + function = app.orchestration_trigger(context_name="ctx", input_type=TypedInput)(orchestrator) + encoded_input = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(TypedInput(7)) + start = [ + helpers.new_orchestrator_started_event(NOW), + helpers.new_execution_started_event("compat_test", INSTANCE_ID, encoded_input), + ] + request = pb.OrchestratorRequest(instanceId=INSTANCE_ID, newEvents=start) + encoded = base64.b64encode(request.SerializeToString()).decode() + result = await function.build().get_user_function()(ctx=encoded) + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(result)) + assert _completed(response) == 7 + + +def test_continue_as_new_shares_compatibility_state() -> None: + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> None: + await ctx.wait_for_external_event("next") + ctx.as_task_context().continue_as_new({"next": 1}) + assert ctx.will_continue_as_new + + result = _execute(orchestrator, _start(), [helpers.new_event_raised_event("next", "null")]) + assert len(result.actions) == 1 + completion = result.actions[0].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert json.loads(completion.result.value) == {"next": 1} + + +@pytest.mark.parametrize("sub_orchestration", [False, True]) +def test_retry_operations_match_generator_replay(sub_orchestration: bool) -> None: + policy = df.RetryPolicy(first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3) + + async def coroutine(ctx: df.AsyncDurableOrchestrationContext) -> Any: + if sub_orchestration: + return await ctx.call_sub_orchestrator_with_retry("child", policy, 3, "child-id") + return await ctx.call_activity_with_retry("work", policy, 3) + + def generator_style(ctx: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, Any]: + if sub_orchestration: + return (yield ctx.call_sub_orchestrator_with_retry("child", policy, 3, "child-id")) + return (yield ctx.call_activity_with_retry("work", policy, 3)) + + def replay(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> pb.OrchestratorResponse: + response = _execute(coroutine, old, new) + assert response.actions == _execute(generator_style, old, new).actions + return response + + start = _start() + initial = replay([], start) + if sub_orchestration: + assert json.loads(initial.actions[0].createSubOrchestration.input.value) == 3 + scheduled = helpers.new_sub_orchestration_created_event(1, "child", "child-id") + failed = helpers.new_sub_orchestration_failed_event(1, ValueError("retry me")) + completed = helpers.new_sub_orchestration_completed_event(1, "42") + else: + assert json.loads(initial.actions[0].scheduleTask.input.value) == 3 + scheduled = helpers.new_task_scheduled_event(1, "work") + failed = helpers.new_task_failed_event(1, ValueError("retry me")) + completed = helpers.new_task_completed_event(1, "42") + history = start + [scheduled] + retry = replay(history, [failed]) + assert retry.actions[0].id == 2 + deadline = retry.actions[0].createTimer.fireAt.ToDatetime() + history += [failed, helpers.new_timer_created_event(2, deadline)] + fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(2, deadline)] + assert replay(history, fired).actions[0].id == 1 + assert _completed(replay(history + fired + [scheduled], [completed])) == 42 + + +def test_entity_call_and_signal_replay_with_v1_argument_names() -> None: + entity = entities.EntityInstanceId("counter", "one") + + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> Any: + ctx.signal_entity(entityId=entity, operationName="add", operationInput=3) + return await ctx.call_entity(entityId=entity, operationName="get") + + start = _start() + initial = _execute(orchestrator, [], start) + assert len(initial.actions) == 2 + signal = initial.actions[0].sendEntityMessage.entityOperationSignaled + call = initial.actions[1].sendEntityMessage.entityOperationCalled + assert signal.operation == "add" + assert json.loads(signal.input.value) == 3 + assert call.operation == "get" + assert call.targetInstanceId.value == str(entity) + history = start + [ + helpers.new_event_sent_event(1, str(entity), json.dumps({"id": signal.requestId})), + helpers.new_event_sent_event(2, str(entity), json.dumps({"id": call.requestId})), + ] + response = helpers.new_event_raised_event(call.requestId, json.dumps({"result": "42"})) + assert _completed(_execute(orchestrator, history, [response])) == 42 + + +def test_durable_http_is_cold_and_decodes_response() -> None: + from azure.durable_functions.http.builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME + + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + invalid = ctx.call_http("POST", "https://example.com", {"bad": True}, is_raw_str=True) + try: + await invalid + except TypeError: + pass + else: + raise AssertionError("Expected invalid raw content to fail when started") + response = await ctx.call_http( + "POST", "https://example.com", {"number": 3}, {"x-test": "value"}, + df.ManagedIdentityTokenSource("https://management.core.windows.net/")) + assert isinstance(response, df.DurableHttpResponse) + return [response.status_code, response.headers, response.content] + + start = _start() + initial = _execute(orchestrator, [], start) + assert len(initial.actions) == 1 + action = initial.actions[0].createSubOrchestration + assert action.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME + request = json.loads(action.input.value) + assert request == { + "method": "POST", + "uri": "https://example.com", + "content": '{"number": 3}', + "headers": {"x-test": "value"}, + "tokenSource": {"kind": "AzureManagedIdentity", "resource": "https://management.core.windows.net/"}, + } + history = start + [helpers.new_sub_orchestration_created_event( + 1, action.name, action.instanceId)] + result = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize( + df.DurableHttpResponse(201, {"reply": "yes"}, "ok")) + assert result is not None + assert _completed(_execute( + orchestrator, history, [helpers.new_sub_orchestration_completed_event(1, result)])) == [ + 201, {"reply": "yes"}, "ok"] + + +def test_durable_http_retries_are_snapshotted_when_coroutine_starts() -> None: + from azure.durable_functions.http.builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME + + class CountingPolicy(df.HttpRetryOptions): + def __init__(self) -> None: + super().__init__( + first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3, + status_codes_to_retry=[503]) + self.snapshots = 0 + + def to_json(self) -> dict[str, Any]: + self.snapshots += 1 + return super().to_json() + + async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str | None: + headers = {"stage": "before"} + policy = CountingPolicy() + request = ctx.call_http("GET", "https://example.com", headers=headers, retry_options=policy) + assert policy.snapshots == 0 + headers["stage"] = "after" + response = await request + assert policy.snapshots > 0 + return response.content + + start = _start() + initial = _execute(orchestrator, [], start) + assert len(initial.actions) == 1 + action = initial.actions[0].createSubOrchestration + assert action.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME + request = json.loads(action.input.value) + assert request["headers"] == {"stage": "after"} + assert request["retryOptions"] == df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3, + status_codes_to_retry=[503]).to_json() + history = start + [helpers.new_sub_orchestration_created_event( + 1, action.name, action.instanceId)] + response = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(df.DurableHttpResponse(200, content="ok")) + assert response is not None + assert _completed(_execute( + orchestrator, history, [helpers.new_sub_orchestration_completed_event(1, response)])) == "ok" diff --git a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py new file mode 100644 index 00000000..3cebe150 --- /dev/null +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -0,0 +1,222 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Coroutine replay through native and compatibility Functions entry points.""" + +import base64 +import asyncio +import json +from collections.abc import Callable +from datetime import datetime, timedelta +from typing import Any + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.orchestrator import Orchestrator +from azure.durable_functions.testing import orchestrator_generator_wrapper +from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) +@pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "v1-handle"]) +@pytest.mark.parametrize("borrowed", [False, True], ids=["coroutines", "borrowed-handles"]) +async def test_coroutine_replay_through_functions_entry_points(native: bool, entry: str, borrowed: bool) -> None: + async def modern(ctx: task.AsyncOrchestrationContext, value: int) -> list[Any]: + if not borrowed: + async with asyncio.TaskGroup() as group: + first = group.create_task(ctx.call_activity("work", input=value, return_type=str)) + second = group.create_task(ctx.call_activity("work", input=value + 1, return_type=str)) + return [first.result(), second.result()] + handles = ctx.as_task_context() + return list(await asyncio.gather( + handles.call_activity("work", input=value, return_type=str), + handles.call_activity("work", input=value + 1, return_type=str), + )) + + async def compatible(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: + value = ctx.get_input() + if not borrowed: + async with asyncio.TaskGroup() as group: + first = group.create_task(ctx.call_activity("work", input_=value)) + second = group.create_task(ctx.call_activity("work", input_=value + 1)) + return [first.result(), second.result()] + handles = ctx.as_task_context() + return list(await asyncio.gather( + handles.call_activity("work", value), + handles.call_activity("work", value + 1), + )) + + fn: Callable[..., Any] = modern if native else compatible + worker = DurableFunctionsWorker() + handle = Orchestrator.create(fn) + app = df.DFApp() + builder = app.orchestration_trigger(context_name="ctx")(fn) + registered = builder.build().get_user_function() + instance_id = "async-functions" + + async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: + encoded = base64.b64encode(request.SerializeToString()).decode() + if entry == "worker-sync": + output = worker.execute_orchestration_request(fn, encoded) + elif entry == "worker-async": + output = await worker.execute_orchestration_request_async(fn, encoded) + elif entry == "decorator": + output = await registered(ctx=encoded) + else: + output = handle(encoded) + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(output)) + return response + + start = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event(fn.__name__, instance_id, "3"), + ] + first = await invoke(pb.OrchestratorRequest(instanceId=instance_id, newEvents=start)) + assert [a.scheduleTask.name for a in first.actions] == ["work", "work"] + assert [json.loads(a.scheduleTask.input.value) for a in first.actions] == [3, 4] + assert [a.id for a in first.actions] == [1, 2] + + response = await invoke(pb.OrchestratorRequest( + instanceId=instance_id, + pastEvents=start + [helpers.new_task_scheduled_event(i, "work") for i in (1, 2)], + newEvents=[ + helpers.new_task_completed_event(2, '"second"'), + helpers.new_task_completed_event(1, '"first"'), + ], + )) + assert len(response.actions) == 1 + complete = response.actions[0].completeOrchestration + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(complete.result.value) == ["first", "second"] + + +@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) +def test_owned_event_timeout_replacement_through_functions_worker(native: bool) -> None: + async def modern(ctx: task.AsyncOrchestrationContext, _: Any) -> str: + try: + return await ctx.wait_for_external_event("approval", timeout=timedelta(seconds=1), data_type=str) + except TimeoutError: + return await ctx.wait_for_external_event("approval", data_type=str) + + async def compatible(ctx: df.AsyncDurableOrchestrationContext) -> str: + try: + return await ctx.wait_for_external_event("approval", str, timeout=timedelta(seconds=1)) + except TimeoutError: + return await ctx.wait_for_external_event("approval", str) + + fn: Callable[..., Any] = modern if native else compatible + worker = DurableFunctionsWorker() + instance_id = "owned-event-functions" + now = datetime(2026, 1, 1) + + def invoke(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> pb.OrchestratorResponse: + request = pb.OrchestratorRequest(instanceId=instance_id, pastEvents=old, newEvents=new) + encoded = base64.b64encode(request.SerializeToString()).decode() + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(worker.execute_orchestration_request(fn, encoded))) + return response + + start = [ + helpers.new_orchestrator_started_event(now), + helpers.new_execution_started_event(fn.__name__, instance_id, "null"), + ] + initial = invoke([], start) + assert len(initial.actions) == 1 + assert initial.actions[0].createTimer.fireAt.ToDatetime() == now + timedelta(seconds=1) + old = start + [helpers.new_timer_created_event(1, now + timedelta(seconds=1))] + timeout = [ + helpers.new_orchestrator_started_event(now + timedelta(seconds=1)), + helpers.new_timer_fired_event(1, now + timedelta(seconds=1)), + ] + assert not invoke(old, timeout).actions + response = invoke(old + timeout, [helpers.new_event_raised_event("approval", '"approved"')]) + assert len(response.actions) == 1 + complete = response.actions[0].completeOrchestration + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails + assert json.loads(complete.result.value) == "approved" + + +@pytest.mark.parametrize("fail", [False, True]) +def test_testing_wrapper_drives_coroutine(fail: bool) -> None: + pending = task.CompletableTask[str]() + + async def orchestrator() -> str: + try: + return await pending + except task.TaskFailedError: + return "caught" + + values = orchestrator_generator_wrapper(orchestrator()) + assert next(values) is pending + if fail: + pending.fail("boom", ValueError("boom")) + else: + pending.complete("done") + assert list(values) == ["caught" if fail else "done"] + + +def test_testing_wrapper_drives_immediate_coroutine() -> None: + async def orchestrator() -> str: + return "done" + + assert list(orchestrator_generator_wrapper(orchestrator())) == ["done"] + + +@pytest.mark.parametrize("fail", [False, True]) +async def test_testing_wrapper_isolates_each_step_from_async_test_loop(fail: bool) -> None: + loop = asyncio.get_running_loop() + caller = asyncio.current_task() + first = task.CompletableTask[str]() + second = task.CompletableTask[str]() + + async def orchestrator() -> str: + assert asyncio._get_running_loop() is None + try: + await first + except task.TaskFailedError: + assert asyncio._get_running_loop() is None + assert asyncio._get_running_loop() is None + return await second + + values = orchestrator_generator_wrapper(orchestrator()) + assert next(values) is first + assert asyncio.get_running_loop() is loop + assert asyncio.current_task() is caller + if fail: + first.fail("boom", ValueError("boom")) + else: + first.complete("first") + assert next(values) is second + assert asyncio.get_running_loop() is loop + assert asyncio.current_task() is caller + second.complete("done") + assert list(values) == ["done"] + assert asyncio.get_running_loop() is loop + assert asyncio.current_task() is caller + await asyncio.sleep(0) + + +@pytest.mark.parametrize("error", [ValueError("test failure"), SystemExit("test exit")]) +async def test_testing_wrapper_restores_async_test_loop_on_exception(error: BaseException) -> None: + loop = asyncio.get_running_loop() + caller = asyncio.current_task() + pending = task.CompletableTask[str]() + + async def orchestrator() -> None: + await pending + raise error + + values = orchestrator_generator_wrapper(orchestrator()) + assert next(values) is pending + pending.complete("done") + with pytest.raises(type(error)) as caught: + next(values) + assert caught.value is error + assert asyncio.get_running_loop() is loop + assert asyncio.current_task() is caller + await asyncio.sleep(0) diff --git a/tests/azure-functions-durable/test_http_retry.py b/tests/azure-functions-durable/test_http_retry.py index 3273bef1..4938500d 100644 --- a/tests/azure-functions-durable/test_http_retry.py +++ b/tests/azure-functions-durable/test_http_retry.py @@ -5,7 +5,9 @@ import io import json import urllib.error +from collections.abc import Generator from datetime import datetime, timedelta, tzinfo +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -361,18 +363,24 @@ def test_excluded_error_and_no_options_return_full_response(with_options): @pytest.mark.parametrize("succeed", [False, True]) -def test_call_http_schedules_snapshot_and_observes_child_result(succeed): +@pytest.mark.parametrize("async_caller", [False, True], ids=["generator", "coroutine"]) +def test_call_http_schedules_snapshot_and_observes_child_result(succeed: bool, async_caller: bool) -> None: codes = [503] policy = df.HttpRetryOptions( first_retry_interval=timedelta(seconds=2), max_number_of_attempts=1, status_codes_to_retry=codes) - def caller(context): + def caller(context: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, df.DurableHttpResponse]: pending = context.call_http("GET", "https://example.com", retry_options=policy) codes.append(404) return (yield pending) - parent = _History(caller) + async def coroutine_caller(context: df.AsyncDurableOrchestrationContext) -> df.DurableHttpResponse: + pending = context.call_http("GET", "https://example.com", retry_options=policy) + codes.append(404) + return await pending + + parent = _History(coroutine_caller if async_caller else caller) action = parent.only_action("createSubOrchestration") assert action.createSubOrchestration.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME payload = json.loads(action.createSubOrchestration.input.value) diff --git a/tests/azure-functions-durable/typing/import_package.py b/tests/azure-functions-durable/typing/import_package.py index 8d69f92d..47d30aa1 100644 --- a/tests/azure-functions-durable/typing/import_package.py +++ b/tests/azure-functions-durable/typing/import_package.py @@ -1,12 +1,16 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. +import asyncio +from collections.abc import Generator from datetime import timedelta -from typing import assert_type +from typing import Any, assert_type import azure.durable_functions as df from azure.durable_functions.client import DurableFunctionsClient, SyncDurableFunctionsClient from azure.durable_functions.worker import DurableFunctionsWorker +from azure.functions.decorators.function_app import FunctionBuilder +from durabletask import task def create_app() -> df.DFApp: @@ -39,3 +43,45 @@ def check_sync_context_managers( async def check_async_context_manager(client: DurableFunctionsClient) -> None: async with client as entered_client: assert_type(entered_client, DurableFunctionsClient) + + +async def check_cold_operations( + context: df.AsyncDurableOrchestrationContext, + native: task.AsyncOrchestrationContext, +) -> None: + pending = asyncio.create_task(native.call_activity("activity", return_type=int)) + assert_type(pending, asyncio.Task[int]) + assert_type(await pending, int) + event = asyncio.create_task(native.wait_for_external_event( + "approval", timeout=timedelta(seconds=1), data_type=str)) + assert_type(await event, str) + assert_type(await context.wait_for_external_event("approval", str), str) + assert_type(context.as_task_context(), df.DurableOrchestrationContext) + assert_type(await context.task_all([pending, pending]), list[int]) + assert_type(await context.task_any([pending]), asyncio.Future[int]) + assert_type(await context.call_http( + "GET", "https://example.com", + retry_options=df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=1), max_number_of_attempts=3), + ), df.DurableHttpResponse) + operation = native.as_task_context().call_activity("shared", return_type=int) + borrower = asyncio.create_task(operation.wait()) + assert_type(borrower, asyncio.Task[int]) + + +def register_orchestrators(app: df.DFApp, worker: DurableFunctionsWorker) -> None: + def generator(context: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, int]: + return (yield context.call_activity("activity")) + + async def compatible(context: df.AsyncDurableOrchestrationContext) -> int: + return await context.call_activity("activity") + + async def native(context: task.AsyncOrchestrationContext, value: int) -> int: + return await context.call_activity("activity", input=value, return_type=int) + + assert_type(app.orchestration_trigger("context")(generator), FunctionBuilder) + assert_type(app.orchestration_trigger("context")(compatible), FunctionBuilder) + assert_type(app.orchestration_trigger("context")(native), FunctionBuilder) + assert_type(worker.execute_orchestration_request(generator, ""), str) + assert_type(worker.execute_orchestration_request(compatible, ""), str) + assert_type(worker.execute_orchestration_request(native, ""), str) From 079744fa1cdc2dc0c5d238d074298b3e04c62fe9 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 12 Nov 2026 17:01:09 -0700 Subject: [PATCH 2/2] Use the native async shape for Functions orchestrators and add call_http helper Remove AsyncDurableOrchestrationContext. Single-parameter async def orchestrators now fail with TypeError when decorated or registered, pointing to the two-parameter AsyncOrchestrationContext shape. Add azure.durable_functions.call_http for durabletask orchestration contexts. It returns a cold coroutine for AsyncOrchestrationContext and an eager task for OrchestrationContext, and shares the existing durable HTTP implementation with DurableOrchestrationContext.call_http. Restore the generator-only orchestrator_generator_wrapper testing helper. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../azure/durable_functions/__init__.py | 8 +- .../decorators/durable_app.py | 9 +- .../azure/durable_functions/http/call.py | 166 +++++ .../internal/compat/orchestration_context.py | 344 ++------- .../azure/durable_functions/orchestrator.py | 8 +- .../testing/orchestrator_generator_wrapper.py | 40 +- .../e2e/apps/dtask_style/orchestrators.py | 26 +- .../e2e/apps/v1_style/orchestrators.py | 23 - .../e2e/test_coroutine_e2e.py | 31 +- .../test_async_compat_context.py | 656 ------------------ .../test_functions_coroutine_orchestrators.py | 172 ++--- .../test_http_retry.py | 89 ++- .../typing/import_package.py | 38 +- 13 files changed, 430 insertions(+), 1180 deletions(-) create mode 100644 azure-functions-durable/azure/durable_functions/http/call.py delete mode 100644 tests/azure-functions-durable/test_async_compat_context.py diff --git a/azure-functions-durable/azure/durable_functions/__init__.py b/azure-functions-durable/azure/durable_functions/__init__.py index 1e4a0fa7..d8406ccb 100644 --- a/azure-functions-durable/azure/durable_functions/__init__.py +++ b/azure-functions-durable/azure/durable_functions/__init__.py @@ -8,6 +8,7 @@ from .decorators.durable_app import Blueprint, DFApp from .client import DurableFunctionsClient, SyncDurableFunctionsClient +from .http.call import call_http from .http.models import DurableHttpRequest, DurableHttpResponse, HttpRetryOptions from .orchestrator import Orchestrator from .internal.compat.retry_options import RetryOptions @@ -17,10 +18,7 @@ from .internal.compat.entity_state_response import EntityStateResponse from .internal.compat.entity_id import EntityId from .internal.compat.token_source import ManagedIdentityTokenSource, TokenSource -from .internal.compat.orchestration_context import ( - AsyncDurableOrchestrationContext, - DurableOrchestrationContext, -) +from .internal.compat.orchestration_context import DurableOrchestrationContext from .internal.compat.entity_context import DurableEntityContext from .internal.compat.compat_aliases import ( DurableOrchestrationClient, @@ -42,7 +40,6 @@ __all__ = [ "Blueprint", - "AsyncDurableOrchestrationContext", "DFApp", "DurableEntityContext", "DurableFunctionsClient", @@ -63,4 +60,5 @@ "RetryOptions", "RetryPolicy", "TokenSource", + "call_http", ] diff --git a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py index be5e8d1d..07f01262 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -21,7 +21,7 @@ builtin_http_poll_orchestrator, ) from ..internal.compat.activity import wrap_activity, wrap_activity_payloads -from ..internal.compat.orchestration_context import FunctionsOrchestrator +from ..internal.compat.orchestration_context import FunctionsOrchestrator, validate_orchestrator from ..internal.invocation import preserve_function_source, wrap_invocation from ..worker import DurableFunctionsWorker @@ -190,6 +190,7 @@ def _configure_orchestrator_callable( """ def decorator(orchestrator_func: FunctionsOrchestrator) -> FunctionBuilder: # Construct an orchestrator based on the end-user code + validate_orchestrator(orchestrator_func) if input_type is not None: # Stash the decorator-declared input type so the runtime can @@ -297,6 +298,12 @@ def orchestration_trigger(self, context_name: str, ``context.get_input()`` decodes the input payload to this type. A call-site ``expected_type`` argument on ``get_input`` takes precedence over this value. + + Raises + ------ + TypeError + When applied to a single-parameter ``async def`` orchestrator. + Async orchestrators take ``(ctx: task.AsyncOrchestrationContext, input)``. """ @self._build_function def wrap(fb: FunctionBuilder) -> FunctionBuilder: diff --git a/azure-functions-durable/azure/durable_functions/http/call.py b/azure-functions-durable/azure/durable_functions/http/call.py new file mode 100644 index 00000000..e4d685f9 --- /dev/null +++ b/azure-functions-durable/azure/durable_functions/http/call.py @@ -0,0 +1,166 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Durable HTTP calls from durabletask orchestration contexts.""" + +from __future__ import annotations + +import json +from collections.abc import Coroutine +from typing import Any, overload + +from durabletask import task + +from ..internal.compat.token_source import TokenSource +from .builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME +from .models import DurableHttpRequest, DurableHttpResponse, HttpRetryOptions + + +def _build_request( + method: str, + uri: str, + content: Any, + headers: dict[str, str] | None, + token_source: TokenSource | None, + is_raw_str: bool, + retry_options: HttpRetryOptions | None) -> dict[str, Any]: + if is_raw_str and not isinstance(content, str): + raise TypeError( + "Invalid use of 'is_raw_str' parameter: 'is_raw_str' is set to " + "'True' but 'content' is not an instance of type 'str'. Either " + "set 'is_raw_str' to 'False', or ensure your 'content' is of " + "type 'str'.") + + json_content: str | None = None + if content is not None: + json_content = content if is_raw_str else json.dumps(content) + + return DurableHttpRequest( + method, uri, json_content, headers, token_source, retry_options).to_json() + + +def schedule_http_request( + ctx: task.OrchestrationContext, + method: str, + uri: str, + content: Any = None, + headers: dict[str, str] | None = None, + token_source: TokenSource | None = None, + is_raw_str: bool = False, + retry_options: HttpRetryOptions | None = None) -> task.Task[DurableHttpResponse]: + """Schedule the built-in durable HTTP sub-orchestration on an eager context.""" + request = _build_request(method, uri, content, headers, token_source, is_raw_str, retry_options) + return ctx.call_sub_orchestrator( + BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, input=request, return_type=DurableHttpResponse) + + +async def _call_http_async( + ctx: task.AsyncOrchestrationContext, + method: str, + uri: str, + content: Any, + headers: dict[str, str] | None, + token_source: TokenSource | None, + is_raw_str: bool, + retry_options: HttpRetryOptions | None) -> DurableHttpResponse: + request = _build_request(method, uri, content, headers, token_source, is_raw_str, retry_options) + return await ctx.call_sub_orchestrator( + BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, input=request, return_type=DurableHttpResponse) + + +@overload +def call_http( + ctx: task.AsyncOrchestrationContext, + method: str, + uri: str, + *, + content: Any = None, + headers: dict[str, str] | None = None, + token_source: TokenSource | None = None, + is_raw_str: bool = False, + retry_options: HttpRetryOptions | None = None, +) -> Coroutine[Any, Any, DurableHttpResponse]: + ... + + +@overload +def call_http( + ctx: task.OrchestrationContext, + method: str, + uri: str, + *, + content: Any = None, + headers: dict[str, str] | None = None, + token_source: TokenSource | None = None, + is_raw_str: bool = False, + retry_options: HttpRetryOptions | None = None, +) -> task.Task[DurableHttpResponse]: + ... + + +def call_http( + ctx: task.AsyncOrchestrationContext | task.OrchestrationContext, + method: str, + uri: str, + *, + content: Any = None, + headers: dict[str, str] | None = None, + token_source: TokenSource | None = None, + is_raw_str: bool = False, + retry_options: HttpRetryOptions | None = None, +) -> Coroutine[Any, Any, DurableHttpResponse] | task.Task[DurableHttpResponse]: + """Make a durable HTTP request from an orchestrator. + + The request is executed by a built-in activity. If the endpoint responds + with ``202 Accepted`` and a ``Location`` header, a built-in + sub-orchestration polls that location until the operation completes, + honoring ``Retry-After``. The result is a :class:`DurableHttpResponse`; + unselected non-2xx responses are returned rather than raised. + + With an :class:`~durabletask.task.AsyncOrchestrationContext`, this returns + a single-use coroutine for ``await``. Like other async context operations, + it is cold: validation, content serialization, and scheduling happen when + the coroutine starts. With an + :class:`~durabletask.task.OrchestrationContext`, the request is scheduled + immediately and a durable task is returned for ``yield``. + + Parameters + ---------- + ctx: AsyncOrchestrationContext | OrchestrationContext + The orchestration context supplied to the orchestrator. + method: str + The HTTP request method. + uri: str + The HTTP request URI. + content: Any + The request content. A ``str`` is sent as JSON unless ``is_raw_str`` + is set; other values are serialized to JSON. + headers: dict[str, str] | None + The request headers. + token_source: TokenSource | None + The source of an OAuth token to add to the request. + is_raw_str: bool + When ``True``, send ``str`` content as-is instead of JSON-encoding + it. Requires ``content`` to be a ``str``. + retry_options: HttpRetryOptions | None + Opt-in durable failure retries for the initial request only. + Selected HTTP statuses and activity failures fail the call after + exhaustion. Location polls do not inherit the policy. Only enable + retries when repeating the request's side effects is safe. + + Raises + ------ + TypeError + If ``ctx`` is not a durabletask orchestration context, or if + ``is_raw_str`` is set and ``content`` is not a ``str``. + """ + if isinstance(ctx, task.AsyncOrchestrationContext): + return _call_http_async( + ctx, method, uri, content, headers, token_source, is_raw_str, retry_options) + # Annotations are not enforced at runtime; reject other contexts clearly. + if not isinstance(ctx, task.OrchestrationContext): # pyright: ignore[reportUnnecessaryIsInstance] + raise TypeError( + "call_http requires a durabletask AsyncOrchestrationContext or OrchestrationContext. " + "Orchestrators that receive a DurableOrchestrationContext should use context.call_http().") + return schedule_http_request( + ctx, method, uri, content, headers, token_source, is_raw_str, retry_options) diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py index ce05f38a..d362ebfc 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py @@ -1,12 +1,9 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -import asyncio import inspect -import json -from collections.abc import Awaitable, Callable, Coroutine, Generator, Sequence -from datetime import UTC, datetime, timedelta -from typing import Any, Optional, TypeAlias, TypeVar, cast, overload +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, TypeAlias, cast from uuid import UUID from durabletask import task @@ -14,12 +11,12 @@ from durabletask.internal.shared import is_async_callable from durabletask.task import OrchestrationContext, RetryPolicy, Task -from ...http.models import DurableHttpResponse, HttpRetryOptions from ..serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER from .function_context import FunctionContext from .token_source import TokenSource -T = TypeVar("T") +if TYPE_CHECKING: + from ...http.models import HttpRetryOptions class DurableOrchestrationContext: @@ -255,294 +252,23 @@ def call_http(self, """ # Imported lazily to avoid a module import cycle # (http.models -> compat.token_source). - from ...http.builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME - from ...http.models import DurableHttpRequest, DurableHttpResponse - - if (not isinstance(content, str)) and is_raw_str: - raise TypeError( - "Invalid use of 'is_raw_str' parameter: 'is_raw_str' is set to " - "'True' but 'content' is not an instance of type 'str'. Either " - "set 'is_raw_str' to 'False', or ensure your 'content' is of " - "type 'str'.") - - json_content: Optional[str] = None - if content is not None: - if isinstance(content, str) and is_raw_str: - json_content = content - else: - json_content = json.dumps(content) - - request = DurableHttpRequest( - method, uri, json_content, headers, token_source, retry_options) - return self._ctx.call_sub_orchestrator( - BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, - input=request.to_json(), - return_type=DurableHttpResponse) - - -def _schedule_awaitables( - operations: Sequence[Awaitable[T]], -) -> tuple[list[asyncio.Future[T]], list[asyncio.Future[T]]]: - """Schedule each distinct awaitable once, retaining input order.""" - scheduled: dict[int, asyncio.Future[T]] = {} - owned: list[asyncio.Future[T]] = [] - futures: list[asyncio.Future[T]] = [] - loop = asyncio.get_running_loop() - for operation in operations: - if not inspect.isawaitable(operation): - raise TypeError("Orchestration operations must be awaitable.") - if isinstance(operation, asyncio.Future) and operation.get_loop() is not loop: - raise ValueError("Orchestration tasks must belong to the current durable event loop.") - for operation in operations: - key = id(operation) - if key not in scheduled: - future = asyncio.ensure_future(operation) - scheduled[key] = future - if future is not operation: - owned.append(future) - futures.append(scheduled[key]) - return futures, owned - - -async def _cancel_owned(futures: Sequence[asyncio.Future[Any]]) -> None: - pending = [future for future in futures if not future.done()] - for future in pending: - future.cancel() - if pending: - await asyncio.gather(*pending, return_exceptions=True) - - -class AsyncDurableOrchestrationContext: - """Functions v1-style context supplied automatically to async orchestrators. - - Durable operations are cold, single-use coroutines. Synchronous commands - and metadata have the same behavior and argument names as the generator - context. Use :meth:`as_task_context` only when eager durable task handles - are needed for interoperability. - """ - - def __init__( - self, ctx: task.AsyncOrchestrationContext, - orchestration_input: Any = None, input_type: type | None = None, - ) -> None: - self._ctx = ctx - self._compat = DurableOrchestrationContext(ctx.as_task_context(), orchestration_input, input_type) - - @overload - def get_input(self, expected_type: type[T]) -> T: - ... - - @overload - def get_input(self, expected_type: None = None) -> Any: - ... - - def get_input(self, expected_type: type | None = None) -> Any: - """Get the orchestration input, optionally coerced to the expected type.""" - return self._compat.get_input(expected_type) + from ...http.call import schedule_http_request - @property - def instance_id(self) -> str: - """Get the current orchestration instance ID.""" - return self._compat.instance_id - - @property - def parent_instance_id(self) -> str | None: - """Get the parent orchestration instance ID, if any.""" - return self._compat.parent_instance_id - - @property - def version(self) -> str | None: - """Get the orchestration instance version, if any.""" - return self._compat.version - - @property - def current_utc_datetime(self) -> datetime: - """Get the replay-safe current UTC time as a timezone-aware datetime.""" - return self._compat.current_utc_datetime - - @property - def is_replaying(self) -> bool: - """Get whether the orchestration is replaying its history.""" - return self._compat.is_replaying - - @property - def custom_status(self) -> Any: - """Get the custom status set during this execution.""" - return self._compat.custom_status - - @property - def will_continue_as_new(self) -> bool: - """Get whether continue-as-new was requested during this execution.""" - return self._compat.will_continue_as_new - - @property - def function_context(self) -> FunctionContext: - """Get the Functions context bag.""" - return self._compat.function_context - - @property - def histories(self) -> Any: - """History is not exposed on orchestration contexts.""" - return self._compat.histories - - def continue_as_new(self, input_: Any) -> None: - """Restart the orchestration with a new input.""" - self._compat.continue_as_new(input_) - - def set_custom_status(self, status: Any) -> None: - """Set the orchestration's custom status.""" - self._compat.set_custom_status(status) - - def new_uuid(self) -> str: - """Create a replay-safe UUID string.""" - return self._compat.new_uuid() - - def new_guid(self) -> UUID: - """Create a replay-safe UUID.""" - return self._compat.new_guid() - - def signal_entity( - self, entityId: EntityInstanceId, operationName: str, operationInput: Any = None, - ) -> None: - """Signal an entity without waiting for a result.""" - self._compat.signal_entity(entityId, operationName, operationInput) - - def as_task_context(self) -> DurableOrchestrationContext: - """Get the v1-compatible eager task-handle context.""" - return self._compat - - async def call_activity(self, name: Callable[..., Any] | str, input_: Any = None) -> Any: - """Start and await an activity.""" - return await self._ctx.call_activity(name, input=input_) - - async def call_activity_with_retry( - self, name: Callable[..., Any] | str, retry_options: RetryPolicy, input_: Any = None, - ) -> Any: - """Start and await an activity with retries.""" - return await self._ctx.call_activity(name, input=input_, retry_policy=retry_options) - - async def call_sub_orchestrator( - self, name: Callable[..., Any] | str, input_: Any = None, instance_id: str | None = None, - ) -> Any: - """Start and await a sub-orchestration.""" - return await self._ctx.call_sub_orchestrator(name, input=input_, instance_id=instance_id) - - async def call_sub_orchestrator_with_retry( - self, name: Callable[..., Any] | str, retry_options: RetryPolicy, - input_: Any = None, instance_id: str | None = None, - ) -> Any: - """Start and await a sub-orchestration with retries.""" - return await self._ctx.call_sub_orchestrator( - name, input=input_, instance_id=instance_id, retry_policy=retry_options) - - async def create_timer(self, fire_at: datetime | timedelta) -> None: - """Wait for a durable timer, withdrawing it locally if cancelled.""" - await self._ctx.create_timer(fire_at) - - @overload - async def wait_for_external_event( - self, name: str, expected_type: type[T], *, timeout: timedelta | None = None, - ) -> T: - ... - - @overload - async def wait_for_external_event( - self, name: str, expected_type: None = None, *, timeout: timedelta | None = None, - ) -> Any: - ... - - async def wait_for_external_event( - self, name: str, expected_type: type[Any] | None = None, *, timeout: timedelta | None = None, - ) -> Any: - """Wait for an event, withdrawing the local subscription on cancellation or timeout.""" - return await self._ctx.wait_for_external_event(name, data_type=expected_type, timeout=timeout) - - async def call_entity( - self, entityId: EntityInstanceId, operationName: str, operationInput: Any = None, - ) -> Any: - """Start and await an entity operation.""" - return await self._ctx.call_entity(entityId, operationName, operationInput) - - async def call_http( - self, method: str, uri: str, content: Any = None, headers: dict[str, str] | None = None, - token_source: TokenSource | None = None, is_raw_str: bool = False, - retry_options: HttpRetryOptions | None = None, - ) -> DurableHttpResponse: - """Start and await a durable HTTP request.""" - return await self._compat.call_http( - method, uri, content, headers, token_source, is_raw_str, retry_options) - - async def task_all(self, operations: Sequence[Awaitable[T]]) -> list[T]: - """Wait for every operation, preserving input order and duplicate references. - - The first observed failure is raised after all operations finish. - Existing asyncio futures are borrowed; cancelling this group leaves - them running. Coroutines started by this group are cancelled and - cleaned up if the group itself is cancelled. - """ - futures, owned = _schedule_awaitables(operations) - unique = list({id(future): future for future in futures}.values()) - failures: list[BaseException] = [] - - def record_failure(future: asyncio.Future[T]) -> None: - if not failures: - if future.cancelled(): - failures.append(asyncio.CancelledError()) - else: - error = future.exception() - if error is not None: - failures.append(error) - - for future in unique: - future.add_done_callback(record_failure) - try: - results = await asyncio.gather(*(asyncio.shield(future) for future in futures), - return_exceptions=True) - if failures: - raise failures[0] - return cast(list[T], results) - except asyncio.CancelledError: - await _cancel_owned(owned) - raise - finally: - for future in unique: - future.remove_done_callback(record_failure) - - async def task_any(self, operations: Sequence[Awaitable[T]]) -> asyncio.Future[T]: - """Return the first completed operation without cancelling the losers. - - The winner is the original asyncio future if one was supplied; a raw - coroutine is wrapped in a new task. Use tasks when identity or later - access to losing operations is needed. Cancelling the group cancels - only coroutines it started, not futures supplied by its caller. - """ - if not operations: - raise ValueError("task_any requires at least one operation.") - futures, owned = _schedule_awaitables(operations) - unique = list({id(future): future for future in futures}.values()) - winner: asyncio.Future[asyncio.Future[T]] = asyncio.get_running_loop().create_future() - - def completed(future: asyncio.Future[T]) -> None: - if not winner.done(): - winner.set_result(future) - - for future in unique: - future.add_done_callback(completed) - try: - return await winner - except asyncio.CancelledError: - await _cancel_owned(owned) - raise - finally: - for future in unique: - future.remove_done_callback(completed) + return schedule_http_request( + self._ctx, method, uri, content, headers, token_source, is_raw_str, retry_options) FunctionsOrchestrator: TypeAlias = ( task.Orchestrator[Any, Any] | Callable[[DurableOrchestrationContext], Any] - | Callable[[AsyncDurableOrchestrationContext], Coroutine[Any, Any, Any]] ) +"""Orchestrator shapes accepted by the Functions decorators and worker. + +Async orchestrators use the durabletask two-argument shape +``async def orchestrator(ctx: task.AsyncOrchestrationContext, input)``. The +single-argument ``DurableOrchestrationContext`` shape is for synchronous and +generator orchestrators only. +""" def accepts_two_positional_args(fn: Callable[..., Any]) -> bool: @@ -576,10 +302,35 @@ def accepts_two_positional_args(fn: Callable[..., Any]) -> bool: return positional >= 2 +def validate_orchestrator(fn: Callable[..., Any]) -> None: + """Reject orchestrator shapes the Functions provider does not support. + + A single-argument ``async def`` orchestrator is rejected because the + ``DurableOrchestrationContext`` API is synchronous. The function is only + inspected, never called, so no coroutine is created. + + Raises + ------ + TypeError + If ``fn`` is a single-argument async orchestrator. + """ + if accepts_two_positional_args(fn) or not is_async_callable(fn): + return + name = getattr(fn, "__name__", "orchestrator") + raise TypeError( + f"Orchestrator '{name}' is an async function with a single context parameter, " + "which is not supported. Declare async orchestrators with two parameters, " + "for example 'async def orchestrator(ctx: task.AsyncOrchestrationContext, input)', " + "where 'task' is 'durabletask.task'. Single-parameter orchestrators that use " + "DurableOrchestrationContext must be generator or regular functions.") + + def wrap_orchestrator(fn: Callable[..., Any]) -> Callable[..., Any]: """Adapt a v1-style one-argument orchestrator to durabletask's ``(ctx, input)``. - Two-argument (durabletask-native) orchestrators are returned unchanged. The + Two-argument (durabletask-native) orchestrators, including ``async def`` + orchestrators, are returned unchanged. Single-argument ``async def`` + orchestrators are rejected with ``TypeError``. The returned wrapper deliberately does not set ``__wrapped__`` so durabletask introspects the wrapper's own ``(context, _input)`` signature rather than the wrapped function's signature. When the ``orchestration_trigger`` decorator @@ -588,26 +339,13 @@ def wrap_orchestrator(fn: Callable[..., Any]) -> Callable[..., Any]: type (required in strict-typing mode, where the codec reconstructs a custom-object payload only when given the destination type). """ + validate_orchestrator(fn) if accepts_two_positional_args(fn): return fn input_type = getattr(fn, "_df_input_type", None) name = getattr(fn, "__name__", "orchestrator") - if is_async_callable(fn): - async def _async_wrapper(context: task.AsyncOrchestrationContext, _input: Any = None) -> Any: - adapter = AsyncDurableOrchestrationContext(context, _input, input_type) - coroutine = fn(adapter) - if not inspect.iscoroutine(coroutine): - if inspect.isgenerator(coroutine): - coroutine.close() - raise TypeError("An async orchestrator must return a coroutine.") - return await coroutine - _async_wrapper.__name__ = name - if input_type is not None: - _async_wrapper.__annotations__["_input"] = input_type - return _async_wrapper - if inspect.isgeneratorfunction(fn): def _generator_wrapper(context: OrchestrationContext, _input: Any = None) -> Any: adapter = DurableOrchestrationContext(context, _input, input_type) diff --git a/azure-functions-durable/azure/durable_functions/orchestrator.py b/azure-functions-durable/azure/durable_functions/orchestrator.py index 0e8e147c..2968b71e 100644 --- a/azure-functions-durable/azure/durable_functions/orchestrator.py +++ b/azure-functions-durable/azure/durable_functions/orchestrator.py @@ -10,6 +10,7 @@ import azure.functions as func +from .internal.compat.orchestration_context import validate_orchestrator from .worker import DurableFunctionsWorker @@ -70,8 +71,13 @@ def create(cls, fn: Callable[..., Any]) -> Callable[[Any], str]: ------- Callable[[Any], str] Handle function of the newly created orchestration client - """ + Raises + ------ + TypeError + If ``fn`` is a single-argument ``async def`` orchestrator. + """ + validate_orchestrator(fn) worker = DurableFunctionsWorker() # The generated handle is the function registered with the Azure diff --git a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py index 491b10c0..43daf0c5 100644 --- a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py +++ b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py @@ -1,18 +1,16 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Replay simulation helper for unit-testing generator or coroutine orchestrators.""" +"""Replay simulation helper for unit-testing orchestrator generators.""" -import asyncio -from collections.abc import Coroutine, Generator -from typing import Any +from typing import Any, Generator def orchestrator_generator_wrapper( - generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]) -> Generator[Any, None, None]: - """Drive a user orchestrator the way the Durable replay engine does. + generator: Generator[Any, Any, Any]) -> Generator[Any, None, None]: + """Drive a user orchestrator generator the way the Durable replay engine does. - A Durable Functions orchestrator yields or awaits durable task objects and + A Durable Functions orchestrator is a generator that yields task objects and is resumed with each task's result until it returns the orchestration output. In a real execution the Durable Task runtime performs this drive loop during replay. This wrapper reproduces that loop in-process so an @@ -22,14 +20,10 @@ def orchestrator_generator_wrapper( back into the orchestrator (mirroring how failed activities, timers, or other tasks surface during replay). - This mock-based helper handles direct durable awaits in synchronous or - async tests. It does not simulate the runtime's asyncio scheduler; use - replay/backend tests for gather, task groups, sleeps and timeouts. - Parameters ---------- - generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any] - The user orchestrator generator or coroutine. It must suspend on tasks + generator: Generator[Any, Any, Any] + The user orchestrator generator. It is expected to yield task objects that expose a ``result`` attribute and to be resumed with those results until it returns the orchestration output. @@ -39,24 +33,14 @@ def orchestrator_generator_wrapper( Each task object yielded by the orchestrator, followed by the orchestrator's final return value. """ - def step(value: Any = None, *, error: Exception | None = None) -> Any: - previous_loop = asyncio._get_running_loop() - try: - if isinstance(generator, Coroutine): - # Direct durable awaits belong to this mock driver, not the - # caller's asyncio loop (or an enclosing durable dispatch). - asyncio._set_running_loop(None) - return generator.throw(error) if error is not None else generator.send(value) - finally: - asyncio._set_running_loop(previous_loop) - try: - previous = step() + previous = next(generator) except StopIteration as e: # An orchestrator that returns immediately without yielding any task # completes right away (mirroring the runtime, which finishes the # orchestration on the first return). Surface its output as the final - # value and stop -- letting StopIteration escape would become RuntimeError. + # value and stop -- ``next`` raising here otherwise leaks as an + # unhandled ``StopIteration``. yield e.value return yield previous @@ -68,9 +52,9 @@ def step(value: Any = None, *, error: Exception | None = None) -> Any: # Simulated activity exceptions, timer-interrupted exceptions, # or any other case where reading a task result would throw: # push the exception back into the orchestrator as replay does. - previous = step(error=e) + previous = generator.throw(e) else: - previous = step(previous_result) + previous = generator.send(previous_result) yield previous except StopIteration as e: # The orchestrator returned; surface its output as the final value. diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 6a6c67ae..3a0cddd8 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -4,7 +4,8 @@ """Orchestrator functions for the durabletask-native-style sample app (blueprint). Every orchestrator uses the modern durabletask two-argument style -(``def orch(ctx, input):``) and the durabletask ``OrchestrationContext`` API +(``def orch(ctx, input):`` or ``async def orch(ctx, input):``) and the +durabletask ``OrchestrationContext`` / ``AsyncOrchestrationContext`` API directly. These cover the inherited durabletask surface -- activity chaining, fan-out/fan-in (``task.when_all``), sub-orchestrations, external events, timers, entities (``ctx.call_entity`` / ``ctx.signal_entity``), custom status, @@ -57,6 +58,29 @@ async def async_cold_probe(ctx: task.AsyncOrchestrationContext, _: Any) -> dict[ return {"greeting": greeting, "total": first.result() + second.result(), "child": child} +@bp.orchestration_trigger(context_name="context") +async def async_fan_out(ctx: task.AsyncOrchestrationContext, count: int | None) -> int: + count = count or 5 + results = await asyncio.gather( + *(ctx.call_activity("square", input=i, return_type=int) for i in range(1, count + 1))) + return sum(results) + + +@bp.orchestration_trigger(context_name="context") +async def async_race(ctx: task.AsyncOrchestrationContext, _: Any) -> dict[str, Any]: + handles = ctx.as_task_context() + activity = handles.call_activity("say_hello", input="Tokyo", return_type=str) + timeout = handles.create_timer(timedelta(minutes=5)) + winner = await task.when_any([activity, timeout]) + return {"activity_won": winner is activity, "result": activity.get_result()} + + +@bp.orchestration_trigger(context_name="context") +async def async_call_http(ctx: task.AsyncOrchestrationContext, url: str) -> dict[str, Any]: + response = await df.call_http(ctx, "GET", url) + return {"status_code": response.status_code, "content": response.content} + + @bp.orchestration_trigger(context_name="context") def logging_filtered(ctx: task.OrchestrationContext, _: Any): _LOGGER.info("worker-info-filter-anchor %s", ctx.instance_id) diff --git a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py index 1477caae..a96467c7 100644 --- a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py @@ -13,8 +13,6 @@ """ from datetime import timedelta -import asyncio -from typing import Any import azure.durable_functions as df @@ -33,27 +31,6 @@ def activity_chain(context: df.DurableOrchestrationContext): return [first, second, third] -@bp.orchestration_trigger(context_name="context") -async def async_await_probe(context: df.AsyncDurableOrchestrationContext) -> dict[str, Any]: - greeting = await context.call_activity("say_hello", "Tokyo") - values = await asyncio.gather(*(context.call_activity("square", i) for i in (2, 3))) - await asyncio.sleep(0) - await asyncio.sleep(0.01) - child = await context.call_sub_orchestrator("activity_chain") - return {"greeting": greeting, "total": sum(values), "child": child} - - -@bp.orchestration_trigger(context_name="context") -async def async_cold_probe(context: df.AsyncDurableOrchestrationContext) -> dict[str, Any]: - greeting = await asyncio.create_task(context.call_activity("say_hello", "Tokyo")) - async with asyncio.TaskGroup() as group: - first = group.create_task(context.call_activity("square", 2)) - second = group.create_task(context.call_activity("square", 3)) - await context.create_timer(timedelta(milliseconds=10)) - child: list[str] = await context.call_sub_orchestrator("activity_chain") - return {"greeting": greeting, "total": first.result() + second.result(), "child": child} - - @bp.orchestration_trigger(context_name="context") def fan_out_fan_in(context: df.DurableOrchestrationContext): count = context.get_input() or 5 diff --git a/tests/azure-functions-durable/e2e/test_coroutine_e2e.py b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py index 6e5f1629..d1832a1b 100644 --- a/tests/azure-functions-durable/e2e/test_coroutine_e2e.py +++ b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py @@ -1,7 +1,9 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Real-host smoke coverage for both async orchestrator authoring styles.""" +"""Real-host smoke coverage for two-argument async orchestrators.""" + +from typing import Any import pytest @@ -11,15 +13,30 @@ pytestmark = pytest.mark.functions_e2e -@pytest.mark.parametrize("app_fixture", ["dtask_app", "v1_app"]) -@pytest.mark.parametrize("orchestrator", ["async_await_probe", "async_cold_probe"]) -def test_async_await_orchestrator(request: pytest.FixtureRequest, app_fixture: str, orchestrator: str) -> None: - app: FunctionApp = request.getfixturevalue(app_fixture) - instance_id = app.start_orchestration(orchestrator) +def _completed_output(app: FunctionApp, name: str, body: Any = None) -> Any: + instance_id = app.start_orchestration(name, body=body) status = app.wait_for_completion(instance_id) assert status["runtimeStatus"].upper() == "COMPLETED", status - assert status["output"] == { + return status["output"] + + +@pytest.mark.parametrize("orchestrator", ["async_await_probe", "async_cold_probe"]) +def test_async_await_orchestrator(dtask_app: FunctionApp, orchestrator: str) -> None: + assert _completed_output(dtask_app, orchestrator) == { "greeting": "Hello Tokyo!", "total": 13, "child": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"], } + + +def test_async_fan_out(dtask_app: FunctionApp) -> None: + assert _completed_output(dtask_app, "async_fan_out", 5) == 55 + + +def test_async_race_returns_winning_handle(dtask_app: FunctionApp) -> None: + assert _completed_output(dtask_app, "async_race") == {"activity_won": True, "result": "Hello Tokyo!"} + + +def test_async_call_http(dtask_app: FunctionApp) -> None: + output = _completed_output(dtask_app, "async_call_http", f"{dtask_app.base_url}/api/ping") + assert output == {"status_code": 200, "content": "pong"} diff --git a/tests/azure-functions-durable/test_async_compat_context.py b/tests/azure-functions-durable/test_async_compat_context.py deleted file mode 100644 index 27b7b6fe..00000000 --- a/tests/azure-functions-durable/test_async_compat_context.py +++ /dev/null @@ -1,656 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - -"""Replay and ownership tests for the Functions coroutine compatibility context.""" - -import asyncio -import base64 -import inspect -import json -from collections.abc import Callable, Coroutine, Generator -from datetime import UTC, datetime, timedelta -from functools import partial, wraps -from typing import Any -from uuid import UUID - -import pytest - -import azure.durable_functions as df -from azure.durable_functions.internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER -from azure.durable_functions.worker import DurableFunctionsWorker -from durabletask import entities, task -from durabletask.internal import helpers -from durabletask.internal import orchestrator_service_pb2 as pb - - -INSTANCE_ID = "async-compat-test" -NOW = datetime(2026, 1, 1) - - -class TypedInput: - def __init__(self, value: int) -> None: - self.value = value - - def to_json(self) -> dict[str, int]: - return {"value": self.value} - - @classmethod - def from_json(cls, data: dict[str, int]) -> "TypedInput": - return cls(data["value"]) - - -def _start() -> list[pb.HistoryEvent]: - return [ - helpers.new_orchestrator_started_event(NOW), - helpers.new_execution_started_event("compat_test", INSTANCE_ID, "null"), - ] - - -def _execute( - orchestrator: Callable[..., Any], - old: list[pb.HistoryEvent], - new: list[pb.HistoryEvent], -) -> pb.OrchestratorResponse: - request = pb.OrchestratorRequest(instanceId=INSTANCE_ID, pastEvents=old, newEvents=new) - encoded = base64.b64encode(request.SerializeToString()).decode() - - def invoke() -> pb.OrchestratorResponse: - response = pb.OrchestratorResponse() - result = DurableFunctionsWorker().execute_orchestration_request(orchestrator, encoded) - response.ParseFromString(base64.b64decode(result)) - return response - - response = invoke() - assert invoke().actions == response.actions - return response - - -def _completed(response: pb.OrchestratorResponse) -> Any: - assert len(response.actions) == 1 - assert response.actions[0].HasField("completeOrchestration") - completed = response.actions[0].completeOrchestration - assert completed.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, completed.failureDetails - return json.loads(completed.result.value) - - -def test_task_all_preserves_duplicates_order_and_shared_tasks() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - shared = asyncio.create_task(ctx.call_activity("shared")) - repeated = ctx.call_activity("repeated") - left = asyncio.create_task(ctx.task_all([repeated, shared, repeated])) - right = asyncio.create_task(ctx.task_all([shared])) - return list(await asyncio.gather(left, right)) - - start = _start() - initial = _execute(orchestrator, [], start) - assert [(action.id, action.scheduleTask.name) for action in initial.actions] == [ - (1, "shared"), (2, "repeated")] - history = start + [ - helpers.new_task_scheduled_event(1, "shared"), - helpers.new_task_scheduled_event(2, "repeated"), - ] - repeated_done = [helpers.new_task_completed_event(2, '"R"')] - assert not _execute(orchestrator, history, repeated_done).actions - shared_done = [helpers.new_task_completed_event(1, '"S"')] - assert _completed(_execute(orchestrator, history + repeated_done, shared_done)) == [ - ["R", "S", "R"], ["S"]] - - -def test_cancelling_one_group_does_not_cancel_another_groups_shared_task() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - shared = asyncio.create_task(ctx.wait_for_external_event("shared")) - cancelled = asyncio.create_task(ctx.task_all([shared])) - survivor = asyncio.create_task(ctx.task_all([shared, shared])) - await asyncio.sleep(0) - await asyncio.sleep(0) - cancelled.cancel() - try: - await cancelled - except asyncio.CancelledError: - pass - else: - raise AssertionError("Expected group cancellation") - return await survivor - - start = _start() - assert not _execute(orchestrator, [], start).actions - assert _completed(_execute( - orchestrator, start, [helpers.new_event_raised_event("shared", '"value"')])) == ["value", "value"] - - -def test_task_all_waits_for_every_child_and_raises_first_observed_failure() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - try: - await ctx.task_all([ - ctx.call_activity("first"), - ctx.call_activity("second"), - ctx.call_activity("last"), - ]) - except task.TaskFailedError as error: - return str(error) - raise AssertionError("Expected the group to fail") - - start = _start() - assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == [ - "first", "second", "last"] - history = start + [ - helpers.new_task_scheduled_event(1, "first"), - helpers.new_task_scheduled_event(2, "second"), - helpers.new_task_scheduled_event(3, "last"), - ] - second_failed = [helpers.new_task_failed_event(2, ValueError("second failure"))] - first_failed = [helpers.new_task_failed_event(1, ValueError("first failure"))] - assert not _execute(orchestrator, history, second_failed).actions - assert not _execute(orchestrator, history + second_failed, first_failed).actions - result = _completed(_execute( - orchestrator, history + second_failed + first_failed, - [helpers.new_task_completed_event(3, '"done"')])) - assert "second failure" in result - assert "first failure" not in result - - -@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) -def test_group_failure_order_follows_completions_in_one_history_batch(group_method: str) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - first = asyncio.create_task(ctx.call_activity("first")) - second = asyncio.create_task(ctx.call_activity("second")) - try: - if group_method == "task_all": - await ctx.task_all([first, second]) - else: - winner = await ctx.task_any([first, second]) - assert winner is second - try: - await winner - finally: - try: - await first - except task.TaskFailedError: - pass - except task.TaskFailedError as error: - return str(error) - raise AssertionError("Expected the group to fail") - - start = _start() - assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["first", "second"] - history = start + [ - helpers.new_task_scheduled_event(1, "first"), - helpers.new_task_scheduled_event(2, "second"), - ] - failures = [ - helpers.new_task_failed_event(2, ValueError("second failure")), - helpers.new_task_failed_event(1, ValueError("first failure")), - ] - result = _completed(_execute(orchestrator, history, failures)) - assert "second failure" in result - assert "first failure" not in result - - -@pytest.mark.parametrize("fail", [False, True]) -def test_task_any_returns_original_winner_and_keeps_loser_running(fail: bool) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - slow = asyncio.create_task(ctx.call_activity("slow")) - fast = asyncio.create_task(ctx.call_activity("fast")) - winner = await ctx.task_any([slow, fast, slow]) - assert winner is fast - assert not slow.cancelled() - try: - value = await winner - except task.TaskFailedError: - value = "failed" - return [value, await slow] - - start = _start() - assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["slow", "fast"] - history = start + [ - helpers.new_task_scheduled_event(1, "slow"), - helpers.new_task_scheduled_event(2, "fast"), - ] - fast_done = [helpers.new_task_failed_event(2, ValueError("fast failure"))] if fail else [ - helpers.new_task_completed_event(2, '"fast"')] - assert not _execute(orchestrator, history, fast_done).actions - slow_done = [helpers.new_task_completed_event(1, '"slow"')] - assert _completed(_execute(orchestrator, history + fast_done, slow_done)) == [ - "failed" if fail else "fast", "slow"] - - -def test_task_any_keeps_owned_coroutine_loser_running() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - async def slower() -> None: - await ctx.call_activity("slow") - await ctx.call_activity("continued") - - winner = await ctx.task_any([slower(), ctx.call_activity("fast")]) - assert await winner == "fast" - return await ctx.wait_for_external_event("finish", str) - - start = _start() - assert [a.scheduleTask.name for a in _execute(orchestrator, [], start).actions] == ["slow", "fast"] - history = start + [ - helpers.new_task_scheduled_event(1, "slow"), - helpers.new_task_scheduled_event(2, "fast"), - ] - fast_done = [helpers.new_task_completed_event(2, '"fast"')] - assert not _execute(orchestrator, history, fast_done).actions - slow_done = [helpers.new_task_completed_event(1, '"slow"')] - continued = _execute(orchestrator, history + fast_done, slow_done) - assert [(a.id, a.scheduleTask.name) for a in continued.actions] == [(3, "continued")] - history += fast_done + slow_done + [ - helpers.new_task_scheduled_event(3, "continued"), - helpers.new_task_completed_event(3, '"done"'), - ] - assert _completed(_execute( - orchestrator, history, [helpers.new_event_raised_event("finish", '"done"')])) == "done" - - -@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) -def test_cancelling_group_withdraws_owned_events_but_not_borrowed_tasks(group_method: str) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - borrowed = asyncio.create_task(ctx.wait_for_external_event("borrowed")) - operations = [ctx.wait_for_external_event("owned"), borrowed] - group = asyncio.create_task(getattr(ctx, group_method)(operations)) - await asyncio.sleep(0) - await asyncio.sleep(0) - group.cancel() - try: - await group - except asyncio.CancelledError: - pass - else: - raise AssertionError("Expected group cancellation") - assert not borrowed.done() - replacement = await ctx.wait_for_external_event("owned") - return [replacement, await borrowed] - - start = _start() - assert not _execute(orchestrator, [], start).actions - events = [ - helpers.new_event_raised_event("owned", '"replacement"'), - helpers.new_event_raised_event("borrowed", '"borrowed"'), - ] - assert _completed(_execute(orchestrator, start, events)) == ["replacement", "borrowed"] - - -@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) -def test_cancelling_group_withdraws_owned_timers_but_not_borrowed_tasks(group_method: str) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - borrowed = asyncio.create_task(ctx.create_timer(timedelta(seconds=5))) - operations = [ctx.create_timer(timedelta(seconds=10)), borrowed] - group = asyncio.create_task(getattr(ctx, group_method)(operations)) - await asyncio.sleep(0) - await asyncio.sleep(0) - group.cancel() - try: - await group - except asyncio.CancelledError: - pass - else: - raise AssertionError("Expected group cancellation") - assert not borrowed.cancelled() - await borrowed - return "survived" - - start = _start() - initial = _execute(orchestrator, [], start) - assert len(initial.actions) == 1 - timer = initial.actions[0] - assert timer.createTimer.fireAt.ToDatetime() == NOW + timedelta(seconds=5) - deadline = NOW + timedelta(seconds=5) - history = start + [helpers.new_timer_created_event(timer.id, deadline)] - fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer.id, deadline)] - assert _completed(_execute(orchestrator, history, fired)) == "survived" - - -@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) -def test_group_timeout_withdraws_owned_event_wait(group_method: str) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - try: - await asyncio.wait_for( - getattr(ctx, group_method)([ctx.wait_for_external_event("approval")]), timeout=1) - except TimeoutError: - return await ctx.wait_for_external_event("approval", str) - raise AssertionError("Expected timeout") - - start = _start() - initial = _execute(orchestrator, [], start) - assert len(initial.actions) == 1 - timer = initial.actions[0] - deadline = NOW + timedelta(seconds=1) - assert timer.createTimer.fireAt.ToDatetime() == deadline - history = start + [helpers.new_timer_created_event(timer.id, deadline)] - timeout = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer.id, deadline)] - assert not _execute(orchestrator, history, timeout).actions - assert _completed(_execute( - orchestrator, history + timeout, [helpers.new_event_raised_event("approval", '"approved"')])) == "approved" - - -def test_empty_and_invalid_groups() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - assert await ctx.task_all([]) == [] - errors: list[str] = [] - invalid: Any = object() - operations: list[Coroutine[Any, Any, Any]] = [ - ctx.task_any([]), ctx.task_all([invalid]), ctx.task_any([invalid])] - for operation in operations: - try: - await operation - except (TypeError, ValueError) as error: - errors.append(str(error)) - return errors - - result = _completed(_execute(orchestrator, [], _start())) - assert result == [ - "task_any requires at least one operation.", - "Orchestration operations must be awaitable.", - "Orchestration operations must be awaitable.", - ] - - -@pytest.mark.parametrize("group_method", ["task_all", "task_any"]) -def test_invalid_group_does_not_start_its_other_operations(group_method: str) -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - operation = ctx.call_activity("work") - try: - await getattr(ctx, group_method)([operation, object()]) - except TypeError as error: - assert str(error) == "Orchestration operations must be awaitable." - else: - raise AssertionError("Expected invalid group to fail") - return await operation - - start = _start() - initial = _execute(orchestrator, [], start) - assert [(a.id, a.scheduleTask.name) for a in initial.actions] == [(1, "work")] - history = start + [helpers.new_task_scheduled_event(1, "work")] - assert _completed(_execute( - orchestrator, history, [helpers.new_task_completed_event(1, '"done"')])) == "done" - - -@pytest.mark.parametrize("shape", ["function", "wrapped", "partial", "object", "partial-object"]) -def test_v1_async_callable_shapes_receive_compatibility_context(shape: str) -> None: - async def plain(ctx: df.AsyncDurableOrchestrationContext) -> Any: - assert isinstance(ctx, df.AsyncDurableOrchestrationContext) - return await ctx.call_activity("work", ctx.get_input()) - - @wraps(plain) - def wrapped(*args: Any, **kwargs: Any) -> Any: - return plain(*args, **kwargs) - - async def with_prefix(prefix: str, ctx: df.AsyncDurableOrchestrationContext) -> Any: - assert isinstance(ctx, df.AsyncDurableOrchestrationContext) - return await ctx.call_activity("work", prefix) - - class AsyncCallable: - async def __call__(self, ctx: df.AsyncDurableOrchestrationContext) -> Any: - return await plain(ctx) - - class PartialCallable: - async def __call__(self, prefix: str, ctx: df.AsyncDurableOrchestrationContext) -> Any: - return await with_prefix(prefix, ctx) - - variants = { - "function": plain, - "wrapped": wrapped, - "partial": partial(with_prefix, "value"), - "object": AsyncCallable(), - "partial-object": partial(PartialCallable(), "value"), - } - fn = variants[shape] - start = _start() - initial = _execute(fn, [], start) - assert len(initial.actions) == 1 - assert initial.actions[0].scheduleTask.name == "work" - history = start + [helpers.new_task_scheduled_event(1, "work")] - assert _completed(_execute(fn, history, [helpers.new_task_completed_event(1, '"done"')])) == "done" - - -def test_unmarked_v1_wrapper_returning_coroutine_fails_and_closes_it() -> None: - created: list[Any] = [] - - async def async_orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str: - return "unexpected" - - def wrapper(ctx: df.DurableOrchestrationContext) -> Any: - coroutine = async_orchestrator(ctx) # type: ignore[arg-type] - created.append(coroutine) - return coroutine - - result = _execute(wrapper, [], _start()) - failure = result.actions[0].completeOrchestration - assert failure.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED - assert "synchronous orchestrator returned a coroutine" in failure.failureDetails.errorMessage - assert all(inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED for coroutine in created) - - -def test_metadata_status_and_eager_context_share_state() -> None: - def inspect_context(ctx: df.DurableOrchestrationContext | df.AsyncDurableOrchestrationContext) -> list[Any]: - ctx.set_custom_status({"step": 1}) - assert ctx.custom_status == {"step": 1} - assert ctx.get_input(dict) == {"number": 7} - assert vars(ctx.function_context) == {} - with pytest.raises(NotImplementedError): - _ = ctx.histories - uuid = ctx.new_uuid() - guid = ctx.new_guid() - assert isinstance(guid, UUID) - return [ - ctx.instance_id, - ctx.parent_instance_id, - ctx.version, - ctx.is_replaying, - ctx.current_utc_datetime.isoformat(), - uuid, - str(guid), - ] - - async def coroutine(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - result = inspect_context(ctx) - eager = ctx.as_task_context() - assert eager.custom_status == ctx.custom_status - eager.set_custom_status({"step": 2}) - assert ctx.custom_status == {"step": 2} - return result - - def generator_style(ctx: df.DurableOrchestrationContext) -> list[Any]: - result = inspect_context(ctx) - ctx.set_custom_status({"step": 2}) - return result - - started = helpers.new_execution_started_event( - "compat_test", INSTANCE_ID, '{"number": 7}', version="v1") - started.executionStarted.parentInstance.CopyFrom(pb.ParentInstanceInfo( - taskScheduledId=1, - orchestrationInstance=pb.OrchestrationInstance(instanceId="parent"))) - start = [helpers.new_orchestrator_started_event(NOW), started] - result = _execute(coroutine, [], start) - assert result.actions == _execute(generator_style, [], start).actions - values = _completed(result) - assert values[:5] == [INSTANCE_ID, "parent", "v1", False, NOW.replace(tzinfo=UTC).isoformat()] - assert str(UUID(values[5])) == values[5] - assert str(UUID(values[6])) == values[6] - assert json.loads(result.customStatus.value) == {"step": 2} - - -@pytest.mark.parametrize("strict", [False, True]) -async def test_decorator_input_type_reaches_async_compat_context(monkeypatch: pytest.MonkeyPatch, strict: bool) -> None: - monkeypatch.setenv("AZURE_FUNCTIONS_DURABLE_STRICT_TYPING", "1" if strict else "0") - app = df.DFApp() - - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> int: - typed = ctx.get_input() - assert isinstance(typed, TypedInput) - assert isinstance(ctx.get_input(TypedInput), TypedInput) - return typed.value - - function = app.orchestration_trigger(context_name="ctx", input_type=TypedInput)(orchestrator) - encoded_input = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(TypedInput(7)) - start = [ - helpers.new_orchestrator_started_event(NOW), - helpers.new_execution_started_event("compat_test", INSTANCE_ID, encoded_input), - ] - request = pb.OrchestratorRequest(instanceId=INSTANCE_ID, newEvents=start) - encoded = base64.b64encode(request.SerializeToString()).decode() - result = await function.build().get_user_function()(ctx=encoded) - response = pb.OrchestratorResponse() - response.ParseFromString(base64.b64decode(result)) - assert _completed(response) == 7 - - -def test_continue_as_new_shares_compatibility_state() -> None: - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> None: - await ctx.wait_for_external_event("next") - ctx.as_task_context().continue_as_new({"next": 1}) - assert ctx.will_continue_as_new - - result = _execute(orchestrator, _start(), [helpers.new_event_raised_event("next", "null")]) - assert len(result.actions) == 1 - completion = result.actions[0].completeOrchestration - assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW - assert json.loads(completion.result.value) == {"next": 1} - - -@pytest.mark.parametrize("sub_orchestration", [False, True]) -def test_retry_operations_match_generator_replay(sub_orchestration: bool) -> None: - policy = df.RetryPolicy(first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3) - - async def coroutine(ctx: df.AsyncDurableOrchestrationContext) -> Any: - if sub_orchestration: - return await ctx.call_sub_orchestrator_with_retry("child", policy, 3, "child-id") - return await ctx.call_activity_with_retry("work", policy, 3) - - def generator_style(ctx: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, Any]: - if sub_orchestration: - return (yield ctx.call_sub_orchestrator_with_retry("child", policy, 3, "child-id")) - return (yield ctx.call_activity_with_retry("work", policy, 3)) - - def replay(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> pb.OrchestratorResponse: - response = _execute(coroutine, old, new) - assert response.actions == _execute(generator_style, old, new).actions - return response - - start = _start() - initial = replay([], start) - if sub_orchestration: - assert json.loads(initial.actions[0].createSubOrchestration.input.value) == 3 - scheduled = helpers.new_sub_orchestration_created_event(1, "child", "child-id") - failed = helpers.new_sub_orchestration_failed_event(1, ValueError("retry me")) - completed = helpers.new_sub_orchestration_completed_event(1, "42") - else: - assert json.loads(initial.actions[0].scheduleTask.input.value) == 3 - scheduled = helpers.new_task_scheduled_event(1, "work") - failed = helpers.new_task_failed_event(1, ValueError("retry me")) - completed = helpers.new_task_completed_event(1, "42") - history = start + [scheduled] - retry = replay(history, [failed]) - assert retry.actions[0].id == 2 - deadline = retry.actions[0].createTimer.fireAt.ToDatetime() - history += [failed, helpers.new_timer_created_event(2, deadline)] - fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(2, deadline)] - assert replay(history, fired).actions[0].id == 1 - assert _completed(replay(history + fired + [scheduled], [completed])) == 42 - - -def test_entity_call_and_signal_replay_with_v1_argument_names() -> None: - entity = entities.EntityInstanceId("counter", "one") - - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> Any: - ctx.signal_entity(entityId=entity, operationName="add", operationInput=3) - return await ctx.call_entity(entityId=entity, operationName="get") - - start = _start() - initial = _execute(orchestrator, [], start) - assert len(initial.actions) == 2 - signal = initial.actions[0].sendEntityMessage.entityOperationSignaled - call = initial.actions[1].sendEntityMessage.entityOperationCalled - assert signal.operation == "add" - assert json.loads(signal.input.value) == 3 - assert call.operation == "get" - assert call.targetInstanceId.value == str(entity) - history = start + [ - helpers.new_event_sent_event(1, str(entity), json.dumps({"id": signal.requestId})), - helpers.new_event_sent_event(2, str(entity), json.dumps({"id": call.requestId})), - ] - response = helpers.new_event_raised_event(call.requestId, json.dumps({"result": "42"})) - assert _completed(_execute(orchestrator, history, [response])) == 42 - - -def test_durable_http_is_cold_and_decodes_response() -> None: - from azure.durable_functions.http.builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME - - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - invalid = ctx.call_http("POST", "https://example.com", {"bad": True}, is_raw_str=True) - try: - await invalid - except TypeError: - pass - else: - raise AssertionError("Expected invalid raw content to fail when started") - response = await ctx.call_http( - "POST", "https://example.com", {"number": 3}, {"x-test": "value"}, - df.ManagedIdentityTokenSource("https://management.core.windows.net/")) - assert isinstance(response, df.DurableHttpResponse) - return [response.status_code, response.headers, response.content] - - start = _start() - initial = _execute(orchestrator, [], start) - assert len(initial.actions) == 1 - action = initial.actions[0].createSubOrchestration - assert action.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME - request = json.loads(action.input.value) - assert request == { - "method": "POST", - "uri": "https://example.com", - "content": '{"number": 3}', - "headers": {"x-test": "value"}, - "tokenSource": {"kind": "AzureManagedIdentity", "resource": "https://management.core.windows.net/"}, - } - history = start + [helpers.new_sub_orchestration_created_event( - 1, action.name, action.instanceId)] - result = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize( - df.DurableHttpResponse(201, {"reply": "yes"}, "ok")) - assert result is not None - assert _completed(_execute( - orchestrator, history, [helpers.new_sub_orchestration_completed_event(1, result)])) == [ - 201, {"reply": "yes"}, "ok"] - - -def test_durable_http_retries_are_snapshotted_when_coroutine_starts() -> None: - from azure.durable_functions.http.builtin import BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME - - class CountingPolicy(df.HttpRetryOptions): - def __init__(self) -> None: - super().__init__( - first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3, - status_codes_to_retry=[503]) - self.snapshots = 0 - - def to_json(self) -> dict[str, Any]: - self.snapshots += 1 - return super().to_json() - - async def orchestrator(ctx: df.AsyncDurableOrchestrationContext) -> str | None: - headers = {"stage": "before"} - policy = CountingPolicy() - request = ctx.call_http("GET", "https://example.com", headers=headers, retry_options=policy) - assert policy.snapshots == 0 - headers["stage"] = "after" - response = await request - assert policy.snapshots > 0 - return response.content - - start = _start() - initial = _execute(orchestrator, [], start) - assert len(initial.actions) == 1 - action = initial.actions[0].createSubOrchestration - assert action.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME - request = json.loads(action.input.value) - assert request["headers"] == {"stage": "after"} - assert request["retryOptions"] == df.HttpRetryOptions( - first_retry_interval=timedelta(seconds=2), max_number_of_attempts=3, - status_codes_to_retry=[503]).to_json() - history = start + [helpers.new_sub_orchestration_created_event( - 1, action.name, action.instanceId)] - response = DEFAULT_FUNCTIONS_DATA_CONVERTER.serialize(df.DurableHttpResponse(200, content="ok")) - assert response is not None - assert _completed(_execute( - orchestrator, history, [helpers.new_sub_orchestration_completed_event(1, response)])) == "ok" diff --git a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py index 3cebe150..f8058ff8 100644 --- a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -1,30 +1,38 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -"""Coroutine replay through native and compatibility Functions entry points.""" +"""Async orchestrator dispatch through the Functions entry points.""" -import base64 import asyncio +import base64 +import gc import json -from collections.abc import Callable +import warnings from datetime import datetime, timedelta from typing import Any import pytest import azure.durable_functions as df +from azure.durable_functions.internal.compat.orchestration_context import wrap_orchestrator from azure.durable_functions.orchestrator import Orchestrator -from azure.durable_functions.testing import orchestrator_generator_wrapper from azure.durable_functions.worker import DurableFunctionsWorker from durabletask import task from durabletask.internal import helpers from durabletask.internal import orchestrator_service_pb2 as pb -@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) -@pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "v1-handle"]) +def _encoded_start(name: str, instance_id: str = "instance") -> str: + request = pb.OrchestratorRequest(instanceId=instance_id, newEvents=[ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event(name, instance_id, "null"), + ]) + return base64.b64encode(request.SerializeToString()).decode() + + +@pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "orchestrator-create"]) @pytest.mark.parametrize("borrowed", [False, True], ids=["coroutines", "borrowed-handles"]) -async def test_coroutine_replay_through_functions_entry_points(native: bool, entry: str, borrowed: bool) -> None: +async def test_coroutine_replay_through_functions_entry_points(entry: str, borrowed: bool) -> None: async def modern(ctx: task.AsyncOrchestrationContext, value: int) -> list[Any]: if not borrowed: async with asyncio.TaskGroup() as group: @@ -37,33 +45,19 @@ async def modern(ctx: task.AsyncOrchestrationContext, value: int) -> list[Any]: handles.call_activity("work", input=value + 1, return_type=str), )) - async def compatible(ctx: df.AsyncDurableOrchestrationContext) -> list[Any]: - value = ctx.get_input() - if not borrowed: - async with asyncio.TaskGroup() as group: - first = group.create_task(ctx.call_activity("work", input_=value)) - second = group.create_task(ctx.call_activity("work", input_=value + 1)) - return [first.result(), second.result()] - handles = ctx.as_task_context() - return list(await asyncio.gather( - handles.call_activity("work", value), - handles.call_activity("work", value + 1), - )) - - fn: Callable[..., Any] = modern if native else compatible worker = DurableFunctionsWorker() - handle = Orchestrator.create(fn) + handle = Orchestrator.create(modern) app = df.DFApp() - builder = app.orchestration_trigger(context_name="ctx")(fn) + builder = app.orchestration_trigger(context_name="ctx")(modern) registered = builder.build().get_user_function() instance_id = "async-functions" async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: encoded = base64.b64encode(request.SerializeToString()).decode() if entry == "worker-sync": - output = worker.execute_orchestration_request(fn, encoded) + output = worker.execute_orchestration_request(modern, encoded) elif entry == "worker-async": - output = await worker.execute_orchestration_request_async(fn, encoded) + output = await worker.execute_orchestration_request_async(modern, encoded) elif entry == "decorator": output = await registered(ctx=encoded) else: @@ -74,7 +68,7 @@ async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: start = [ helpers.new_orchestrator_started_event(), - helpers.new_execution_started_event(fn.__name__, instance_id, "3"), + helpers.new_execution_started_event(modern.__name__, instance_id, "3"), ] first = await invoke(pb.OrchestratorRequest(instanceId=instance_id, newEvents=start)) assert [a.scheduleTask.name for a in first.actions] == ["work", "work"] @@ -95,21 +89,13 @@ async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: assert json.loads(complete.result.value) == ["first", "second"] -@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) -def test_owned_event_timeout_replacement_through_functions_worker(native: bool) -> None: +def test_owned_event_timeout_replacement_through_functions_worker() -> None: async def modern(ctx: task.AsyncOrchestrationContext, _: Any) -> str: try: return await ctx.wait_for_external_event("approval", timeout=timedelta(seconds=1), data_type=str) except TimeoutError: return await ctx.wait_for_external_event("approval", data_type=str) - async def compatible(ctx: df.AsyncDurableOrchestrationContext) -> str: - try: - return await ctx.wait_for_external_event("approval", str, timeout=timedelta(seconds=1)) - except TimeoutError: - return await ctx.wait_for_external_event("approval", str) - - fn: Callable[..., Any] = modern if native else compatible worker = DurableFunctionsWorker() instance_id = "owned-event-functions" now = datetime(2026, 1, 1) @@ -118,12 +104,12 @@ def invoke(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> pb.Orchest request = pb.OrchestratorRequest(instanceId=instance_id, pastEvents=old, newEvents=new) encoded = base64.b64encode(request.SerializeToString()).decode() response = pb.OrchestratorResponse() - response.ParseFromString(base64.b64decode(worker.execute_orchestration_request(fn, encoded))) + response.ParseFromString(base64.b64decode(worker.execute_orchestration_request(modern, encoded))) return response start = [ helpers.new_orchestrator_started_event(now), - helpers.new_execution_started_event(fn.__name__, instance_id, "null"), + helpers.new_execution_started_event(modern.__name__, instance_id, "null"), ] initial = invoke([], start) assert len(initial.actions) == 1 @@ -141,82 +127,38 @@ def invoke(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]) -> pb.Orchest assert json.loads(complete.result.value) == "approved" -@pytest.mark.parametrize("fail", [False, True]) -def test_testing_wrapper_drives_coroutine(fail: bool) -> None: - pending = task.CompletableTask[str]() - - async def orchestrator() -> str: - try: - return await pending - except task.TaskFailedError: - return "caught" - - values = orchestrator_generator_wrapper(orchestrator()) - assert next(values) is pending - if fail: - pending.fail("boom", ValueError("boom")) - else: - pending.complete("done") - assert list(values) == ["caught" if fail else "done"] - +def test_two_argument_async_orchestrator_passes_through() -> None: + async def native(ctx: task.AsyncOrchestrationContext, value: int) -> int: + return await ctx.call_activity("work", input=value, return_type=int) -def test_testing_wrapper_drives_immediate_coroutine() -> None: - async def orchestrator() -> str: - return "done" - - assert list(orchestrator_generator_wrapper(orchestrator())) == ["done"] - - -@pytest.mark.parametrize("fail", [False, True]) -async def test_testing_wrapper_isolates_each_step_from_async_test_loop(fail: bool) -> None: - loop = asyncio.get_running_loop() - caller = asyncio.current_task() - first = task.CompletableTask[str]() - second = task.CompletableTask[str]() - - async def orchestrator() -> str: - assert asyncio._get_running_loop() is None - try: - await first - except task.TaskFailedError: - assert asyncio._get_running_loop() is None - assert asyncio._get_running_loop() is None - return await second - - values = orchestrator_generator_wrapper(orchestrator()) - assert next(values) is first - assert asyncio.get_running_loop() is loop - assert asyncio.current_task() is caller - if fail: - first.fail("boom", ValueError("boom")) - else: - first.complete("first") - assert next(values) is second - assert asyncio.get_running_loop() is loop - assert asyncio.current_task() is caller - second.complete("done") - assert list(values) == ["done"] - assert asyncio.get_running_loop() is loop - assert asyncio.current_task() is caller - await asyncio.sleep(0) - - -@pytest.mark.parametrize("error", [ValueError("test failure"), SystemExit("test exit")]) -async def test_testing_wrapper_restores_async_test_loop_on_exception(error: BaseException) -> None: - loop = asyncio.get_running_loop() - caller = asyncio.current_task() - pending = task.CompletableTask[str]() - - async def orchestrator() -> None: - await pending - raise error - - values = orchestrator_generator_wrapper(orchestrator()) - assert next(values) is pending - pending.complete("done") - with pytest.raises(type(error)) as caught: - next(values) - assert caught.value is error - assert asyncio.get_running_loop() is loop - assert asyncio.current_task() is caller - await asyncio.sleep(0) + assert wrap_orchestrator(native) is native + builder = df.DFApp().orchestration_trigger(context_name="ctx", input_type=int)(native) + registered = builder.build().get_user_function() + assert registered.orchestrator_function is native + + +@pytest.mark.parametrize("entry", ["decorator", "orchestrator-create", "worker-sync", "worker-async"]) +async def test_single_argument_async_orchestrator_is_rejected(entry: str) -> None: + called = False + + async def legacy(context: df.DurableOrchestrationContext) -> str: + nonlocal called + called = True + return "unreachable" + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + with pytest.raises(TypeError, match=r"legacy.*async def orchestrator\(ctx: task\.AsyncOrchestrationContext"): + if entry == "decorator": + df.DFApp().orchestration_trigger(context_name="context")(legacy) + elif entry == "orchestrator-create": + Orchestrator.create(legacy) + elif entry == "worker-sync": + DurableFunctionsWorker().execute_orchestration_request(legacy, _encoded_start("legacy")) + else: + await DurableFunctionsWorker().execute_orchestration_request_async( + legacy, _encoded_start("legacy")) + gc.collect() + + assert not called + assert [str(w.message) for w in caught if issubclass(w.category, RuntimeWarning)] == [] diff --git a/tests/azure-functions-durable/test_http_retry.py b/tests/azure-functions-durable/test_http_retry.py index 4938500d..138c3aa2 100644 --- a/tests/azure-functions-durable/test_http_retry.py +++ b/tests/azure-functions-durable/test_http_retry.py @@ -362,25 +362,40 @@ def test_excluded_error_and_no_options_return_full_response(with_options): assert response.to_json() == {"status_code": 503, "headers": {"X-Error": "detail"}, "content": "body"} -@pytest.mark.parametrize("succeed", [False, True]) -@pytest.mark.parametrize("async_caller", [False, True], ids=["generator", "coroutine"]) -def test_call_http_schedules_snapshot_and_observes_child_result(succeed: bool, async_caller: bool) -> None: - codes = [503] - policy = df.HttpRetryOptions( - first_retry_interval=timedelta(seconds=2), - max_number_of_attempts=1, status_codes_to_retry=codes) - +def _v1_caller(policy: df.HttpRetryOptions, codes: list[int]): def caller(context: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, df.DurableHttpResponse]: pending = context.call_http("GET", "https://example.com", retry_options=policy) codes.append(404) return (yield pending) + return caller - async def coroutine_caller(context: df.AsyncDurableOrchestrationContext) -> df.DurableHttpResponse: - pending = context.call_http("GET", "https://example.com", retry_options=policy) + +def _task_caller(policy: df.HttpRetryOptions, codes: list[int]): + def caller(ctx: task.OrchestrationContext, _: Any) -> Generator[task.Task[Any], Any, df.DurableHttpResponse]: + pending = df.call_http(ctx, "GET", "https://example.com", retry_options=policy) + codes.append(404) + return (yield pending) + return caller + + +def _async_caller(policy: df.HttpRetryOptions, codes: list[int]): + async def caller(ctx: task.AsyncOrchestrationContext, _: Any) -> df.DurableHttpResponse: + pending = df.call_http(ctx, "GET", "https://example.com", retry_options=policy) codes.append(404) return await pending + return caller + + +@pytest.mark.parametrize("outcome", ["success", "polled", "failure"]) +@pytest.mark.parametrize("make_caller", [_v1_caller, _task_caller, _async_caller], + ids=["v1-context", "task-context", "async-context"]) +def test_call_http_schedules_snapshot_and_observes_child_result(make_caller, outcome: str) -> None: + codes = [503] + policy = df.HttpRetryOptions( + first_retry_interval=timedelta(seconds=2), + max_number_of_attempts=1, status_codes_to_retry=codes) - parent = _History(coroutine_caller if async_caller else caller) + parent = _History(make_caller(policy, codes)) action = parent.only_action("createSubOrchestration") assert action.createSubOrchestration.name == BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME payload = json.loads(action.createSubOrchestration.input.value) @@ -389,11 +404,17 @@ async def coroutine_caller(context: df.AsyncDurableOrchestrationContext) -> df.D action.id, BUILTIN_HTTP_POLL_ORCHESTRATOR_NAME, "child", encoded_input=action.createSubOrchestration.input.value)) child = _History(builtin_http_poll_orchestrator, payload, parent="instance") - child.run_activity(status=200 if succeed else 503) + if outcome == "polled": + child.run_activity(status=202, headers={"Location": "/status", "Retry-After": "3"}) + child.fire_timer(timedelta(seconds=3)) + assert child.run_activity()["uri"] == "https://example.com/status" + else: + child.run_activity(status=200 if outcome == "success" else 503) completion = child.only_action("completeOrchestration").completeOrchestration - if succeed: + if outcome != "failure": parent.dispatch([helpers.new_sub_orchestration_completed_event(action.id, completion.result.value)]) assert parent.result().status_code == 200 + assert parent.result().content == "body" else: failure = helpers.new_sub_orchestration_failed_event(action.id, RuntimeError("placeholder")) failure.subOrchestrationInstanceFailed.failureDetails.CopyFrom(completion.failureDetails) @@ -401,3 +422,45 @@ async def coroutine_caller(context: df.AsyncDurableOrchestrationContext) -> df.D completion = parent.only_action("completeOrchestration").completeOrchestration assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED assert "status code 503" in str(completion.failureDetails) + + +def test_async_call_http_is_cold_until_awaited() -> None: + async def caller(ctx: task.AsyncOrchestrationContext, _: Any) -> str: + pending = df.call_http(ctx, "POST", "https://example.com", content=1, is_raw_str=True) + await ctx.create_timer(timedelta(seconds=1)) + with pytest.raises(TypeError, match="is_raw_str"): + await pending + return "validated" + + history = _History(caller) + # Neither validation nor scheduling happens when the coroutine is created. + history.only_action("createTimer") + history.fire_timer(timedelta(seconds=1)) + completion = history.only_action("completeOrchestration").completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, completion.failureDetails + assert json.loads(completion.result.value) == "validated" + + +@pytest.mark.parametrize(("content", "is_raw_str", "expected"), [ + (None, False, None), + ({"a": 1}, False, '{"a": 1}'), + ("text", False, '"text"'), + ("text", True, "text"), +]) +def test_call_http_content_encoding_matches_v1_context(content: Any, is_raw_str: bool, expected: str | None) -> None: + async def caller(ctx: task.AsyncOrchestrationContext, _: Any) -> df.DurableHttpResponse: + return await df.call_http( + ctx, "POST", "https://example.com", content=content, is_raw_str=is_raw_str, + headers={"X-Test": "value"}) + + action = _History(caller).only_action("createSubOrchestration") + payload = json.loads(action.createSubOrchestration.input.value) + assert payload.get("content") == expected + assert payload["headers"] == {"X-Test": "value"} + assert "retryOptions" not in payload + + +def test_call_http_rejects_v1_context() -> None: + context = df.DurableOrchestrationContext(MagicMock(spec=task.OrchestrationContext)) + with pytest.raises(TypeError, match=r"context\.call_http"): + df.call_http(context, "GET", "https://example.com") # type: ignore[call-overload] diff --git a/tests/azure-functions-durable/typing/import_package.py b/tests/azure-functions-durable/typing/import_package.py index 47d30aa1..f2588306 100644 --- a/tests/azure-functions-durable/typing/import_package.py +++ b/tests/azure-functions-durable/typing/import_package.py @@ -1,8 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -import asyncio -from collections.abc import Generator +from collections.abc import Coroutine, Generator from datetime import timedelta from typing import Any, assert_type @@ -45,43 +44,28 @@ async def check_async_context_manager(client: DurableFunctionsClient) -> None: assert_type(entered_client, DurableFunctionsClient) -async def check_cold_operations( - context: df.AsyncDurableOrchestrationContext, +async def check_call_http( native: task.AsyncOrchestrationContext, + eager: task.OrchestrationContext, + context: df.DurableOrchestrationContext, ) -> None: - pending = asyncio.create_task(native.call_activity("activity", return_type=int)) - assert_type(pending, asyncio.Task[int]) - assert_type(await pending, int) - event = asyncio.create_task(native.wait_for_external_event( - "approval", timeout=timedelta(seconds=1), data_type=str)) - assert_type(await event, str) - assert_type(await context.wait_for_external_event("approval", str), str) - assert_type(context.as_task_context(), df.DurableOrchestrationContext) - assert_type(await context.task_all([pending, pending]), list[int]) - assert_type(await context.task_any([pending]), asyncio.Future[int]) - assert_type(await context.call_http( - "GET", "https://example.com", - retry_options=df.HttpRetryOptions( - first_retry_interval=timedelta(seconds=1), max_number_of_attempts=3), - ), df.DurableHttpResponse) - operation = native.as_task_context().call_activity("shared", return_type=int) - borrower = asyncio.create_task(operation.wait()) - assert_type(borrower, asyncio.Task[int]) + retry = df.HttpRetryOptions(first_retry_interval=timedelta(seconds=1), max_number_of_attempts=3) + pending = assert_type( + df.call_http(native, "GET", "https://example.com", retry_options=retry), + Coroutine[Any, Any, df.DurableHttpResponse]) + assert_type(await pending, df.DurableHttpResponse) + assert_type(df.call_http(eager, "GET", "https://example.com"), task.Task[df.DurableHttpResponse]) + assert_type(context.call_http("GET", "https://example.com"), task.Task[Any]) def register_orchestrators(app: df.DFApp, worker: DurableFunctionsWorker) -> None: def generator(context: df.DurableOrchestrationContext) -> Generator[task.Task[Any], Any, int]: return (yield context.call_activity("activity")) - async def compatible(context: df.AsyncDurableOrchestrationContext) -> int: - return await context.call_activity("activity") - async def native(context: task.AsyncOrchestrationContext, value: int) -> int: return await context.call_activity("activity", input=value, return_type=int) assert_type(app.orchestration_trigger("context")(generator), FunctionBuilder) - assert_type(app.orchestration_trigger("context")(compatible), FunctionBuilder) assert_type(app.orchestration_trigger("context")(native), FunctionBuilder) assert_type(worker.execute_orchestration_request(generator, ""), str) - assert_type(worker.execute_orchestration_request(compatible, ""), str) assert_type(worker.execute_orchestration_request(native, ""), str)