[Async stack 5/6] Functions async orchestration support

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",
 ]

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:

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)

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
 

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