diff --git a/azure-functions-durable/azure/durable_functions/__init__.py b/azure-functions-durable/azure/durable_functions/__init__.py index de861727..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 @@ -59,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 4d7df348..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,6 +21,7 @@ builtin_http_poll_orchestrator, ) from ..internal.compat.activity import wrap_activity, wrap_activity_payloads +from ..internal.compat.orchestration_context import FunctionsOrchestrator, validate_orchestrator 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,8 +188,9 @@ 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 + validate_orchestrator(orchestrator_func) if input_type is not None: # Stash the decorator-declared input type so the runtime can @@ -281,7 +283,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 @@ -296,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 84561e39..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 @@ -2,13 +2,13 @@ # Licensed under the MIT License. import inspect -import json from datetime import UTC, datetime -from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, cast +from typing import TYPE_CHECKING, Any, Callable, Generator, Optional, TypeAlias, cast 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 ..serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER @@ -252,29 +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) + from ...http.call import schedule_http_request + + 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] +) +"""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: @@ -308,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 @@ -320,6 +339,7 @@ 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 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/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..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, @@ -13,6 +14,7 @@ """ from datetime import timedelta +import asyncio import logging from typing import Any @@ -35,6 +37,50 @@ 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") +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/test_coroutine_e2e.py b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py new file mode 100644 index 00000000..d1832a1b --- /dev/null +++ b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py @@ -0,0 +1,42 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Real-host smoke coverage for two-argument async orchestrators.""" + +from typing import Any + +import pytest + +from ._harness import FunctionApp + + +pytestmark = pytest.mark.functions_e2e + + +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 + 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_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py new file mode 100644 index 00000000..f8058ff8 --- /dev/null +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -0,0 +1,164 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Async orchestrator dispatch through the Functions entry points.""" + +import asyncio +import base64 +import gc +import json +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.worker import DurableFunctionsWorker +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +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(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), + )) + + worker = DurableFunctionsWorker() + handle = Orchestrator.create(modern) + app = df.DFApp() + 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(modern, encoded) + elif entry == "worker-async": + output = await worker.execute_orchestration_request_async(modern, 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(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"] + 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"] + + +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) + + 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(modern, encoded))) + return response + + start = [ + helpers.new_orchestrator_started_event(now), + helpers.new_execution_started_event(modern.__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" + + +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) + + 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 3273bef1..138c3aa2 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 @@ -360,19 +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]) -def test_call_http_schedules_snapshot_and_observes_child_result(succeed): +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 + + +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) - def caller(context): - pending = context.call_http("GET", "https://example.com", retry_options=policy) - codes.append(404) - return (yield pending) - - parent = _History(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) @@ -381,11 +404,17 @@ def caller(context): 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) @@ -393,3 +422,45 @@ def caller(context): 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 8d69f92d..f2588306 100644 --- a/tests/azure-functions-durable/typing/import_package.py +++ b/tests/azure-functions-durable/typing/import_package.py @@ -1,12 +1,15 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. +from collections.abc import Coroutine, 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 +42,30 @@ 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_call_http( + native: task.AsyncOrchestrationContext, + eager: task.OrchestrationContext, + context: df.DurableOrchestrationContext, +) -> None: + 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 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")(native), FunctionBuilder) + assert_type(worker.execute_orchestration_request(generator, ""), str) + assert_type(worker.execute_orchestration_request(native, ""), str)