Add sync and async client parity

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

Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0

Author: andystaples

CHANGELOG.md

@@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ADDED
 
+- Added asynchronous `AsyncScheduledTaskClient` / `AsyncScheduleClient` and
+  `AsyncExportHistoryClient` / `AsyncExportHistoryJobClient` APIs. Applications
+  using `AsyncTaskHubGrpcClient` can now manage scheduled tasks and history
+  export jobs without a synchronous client bridge.
 - Added `FailureDetails.is_caused_by()` to check whether a task failure was caused by a given exception type, mirroring .NET's `TaskFailureDetails.IsCausedBy<T>()` and Java's `FailureDetails.isCausedBy()`. Passing an exception type performs a base-type-aware match (a failure caused by a subclass matches its base type) against exception classes already imported in the process; passing a string matches by qualified or unqualified name.
 - Added a per-invocation dependency-resolution API to `durabletask.extensions.history_export` so any hosting model can supply the export client and writer without a process-global. The export activities now resolve their `HistoryExportContext` from a resolver invoked once per activity execution; new public building blocks are the `HistoryExportContextResolver` type, the pure activity bodies `run_list_terminal_instances(context, input)` and `run_export_instance_history(context, input)`, and the `build_activities(resolver)` factory. This lets host-driven, multi-process models (such as Azure Functions, where the process that starts an export job is not the worker that runs an export activity) inject dependencies lazily per invocation.
 

azure-functions-durable/CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+ADDED
+
+- Added `SyncDurableFunctionsClient` and
+  `DFApp.durable_client_input_sync()` for synchronous durable-client functions.
+  Both synchronous and asynchronous Functions durable clients can now use the
+  scheduled-tasks and history-export client APIs without an async-to-sync bridge.
+
 ## 2.0.0b1
 
 First preview (beta) release of `azure-functions-durable` 2.x — a ground-up

azure-functions-durable/azure/durable_functions/__init__.py

@@ -7,7 +7,7 @@
 from durabletask.task import RetryPolicy
 
 from .decorators.durable_app import Blueprint, DFApp
-from .client import DurableFunctionsClient
+from .client import DurableFunctionsClient, SyncDurableFunctionsClient
 from .http.models import DurableHttpRequest, DurableHttpResponse
 from .orchestrator import Orchestrator
 from .internal.compat.retry_options import RetryOptions
@@ -42,6 +42,7 @@
     "DFApp",
     "DurableEntityContext",
     "DurableFunctionsClient",
+    "SyncDurableFunctionsClient",
     "DurableHttpRequest",
     "DurableHttpResponse",
     "DurableOrchestrationClient",

azure-functions-durable/azure/durable_functions/client.py

@@ -14,10 +14,14 @@
     AsyncTaskHubGrpcClient,
     OrchestrationQuery,
     OrchestrationStatus,
+    TaskHubGrpcClient,
 )
 from durabletask.entities import EntityInstanceId
 from durabletask.grpc_options import GrpcChannelOptions
-from .internal.azurefunctions_grpc_interceptor import AzureFunctionsAsyncDefaultClientInterceptorImpl
+from .internal.azurefunctions_grpc_interceptor import (
+    AzureFunctionsAsyncDefaultClientInterceptorImpl,
+    AzureFunctionsDefaultClientInterceptorImpl,
+)
 from .internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER
 from .http import HttpManagementPayload
 from .internal.compat.durable_orchestration_status import DurableOrchestrationStatus
@@ -490,3 +494,30 @@ def _create_http_response(status_code: int, body: Union[str, Any]) -> func.HttpR
             body=body_as_json,
             mimetype="application/json",
             headers={"Content-Type": "application/json"})
+
+
+class SyncDurableFunctionsClient(TaskHubGrpcClient):
+    """Synchronous durable client supplied by a Functions durable-client binding."""
+
+    def __init__(self, client_as_string: str):
+        self._parse_client_configuration(client_as_string)
+        interceptors = [AzureFunctionsDefaultClientInterceptorImpl(
+            self.taskHubName, self.requiredQueryStringParameters)]
+        channel_options: GrpcChannelOptions | None = None
+        if self.maxGrpcMessageSizeInBytes > 0:
+            channel_options = GrpcChannelOptions(
+                max_receive_message_length=self.maxGrpcMessageSizeInBytes,
+                max_send_message_length=self.maxGrpcMessageSizeInBytes)
+        super().__init__(
+            host_address=self.rpcBaseUrl,
+            secure_channel=False,
+            metadata=None,
+            interceptors=interceptors,
+            channel_options=channel_options,
+            data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER)
+
+    _parse_client_configuration = DurableFunctionsClient._parse_client_configuration
+    create_check_status_response = DurableFunctionsClient.create_check_status_response
+    create_http_management_payload = DurableFunctionsClient.create_http_management_payload
+    _get_client_response_links = DurableFunctionsClient._get_client_response_links
+    _get_instance_status_url = DurableFunctionsClient._get_instance_status_url

azure-functions-durable/azure/durable_functions/decorators/durable_app.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import inspect
+from functools import wraps
 from typing import Any, Callable, Optional, Union
 
 import azure.functions as func
@@ -150,12 +152,12 @@ def configure_history_export(self, writer: Any) -> None:
         # whatever worker runs them). ``durable_client_input`` is applied as the
         # outer decorator over ``activity_trigger`` so the built function carries
         # both bindings.
-        self.durable_client_input(client_name="client")(
+        self.durable_client_input(client_name="client", sync=True)(
             self.activity_trigger(
                 input_name="input",
                 activity=LIST_TERMINAL_INSTANCES_ACTIVITY)(
                     list_terminal_instances_client_bound))
-        self.durable_client_input(client_name="client")(
+        self.durable_client_input(client_name="client", sync=True)(
             self.activity_trigger(
                 input_name="input",
                 activity=EXPORT_INSTANCE_HISTORY_ACTIVITY)(
@@ -356,7 +358,9 @@ def decorator() -> FunctionBuilder:
     def durable_client_input(self,
                              client_name: str,
                              task_hub: Optional[str] = None,
-                             connection_name: Optional[str] = None
+                             connection_name: Optional[str] = None,
+                             *,
+                             sync: bool = False,
                              ) -> Callable[[Callable[..., Any]], FunctionBuilder]:
         """Register a Durable-client Function.
 
@@ -374,6 +378,9 @@ def durable_client_input(self,
             The storage account represented by this connection string must be the same one
             used by the target orchestrator functions. If not specified, the default storage
             account connection string for the function app is used.
+        sync: bool
+            When ``True``, inject a :class:`SyncDurableFunctionsClient`. The
+            default injects the asynchronous :class:`DurableFunctionsClient`.
         """
 
         @self._build_function
@@ -400,12 +407,50 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder:
             # user function is registered directly and ``.client_function``
             # points back at it. Internal callers pass an already-built
             # ``FunctionBuilder`` (e.g. history export), which is left untouched.
-            if not isinstance(user_fn, FunctionBuilder):
-                user_fn.client_function = user_fn  # pyright: ignore[reportFunctionMemberAccess]
-            return wrap(user_fn)
+            from ..client import DurableFunctionsClient, SyncDurableFunctionsClient
+
+            function = (user_fn._function._func if isinstance(user_fn, FunctionBuilder)
+                        else user_fn)
+            signature = inspect.signature(function)
+
+            @wraps(function)
+            async def client_bound(*args: Any, **kwargs: Any) -> Any:
+                bound = signature.bind(*args, **kwargs)
+                raw_client = bound.arguments[client_name]
+                if not isinstance(raw_client, str):
+                    raise TypeError(
+                        f"durable client binding '{client_name}' did not provide its configuration")
+                client = (SyncDurableFunctionsClient(raw_client) if sync
+                          else DurableFunctionsClient(raw_client))
+                bound.arguments[client_name] = client
+                try:
+                    result = function(*bound.args, **bound.kwargs)
+                    return await result if inspect.isawaitable(result) else result
+                finally:
+                    if sync:
+                        client.close()
+                    else:
+                        client.schedule_close()
+
+            client_bound.__annotations__[client_name] = str
+            client_bound.client_function = function  # pyright: ignore[reportFunctionMemberAccess]
+            if isinstance(user_fn, FunctionBuilder):
+                user_fn._function._func = client_bound
+                return wrap(user_fn)
+            return wrap(client_bound)
 
         return attach_client_function
 
+    def durable_client_input_sync(
+            self,
+            client_name: str,
+            task_hub: Optional[str] = None,
+            connection_name: Optional[str] = None,
+    ) -> Callable[[Callable[..., Any]], FunctionBuilder]:
+        """Register a durable-client binding that injects a synchronous client."""
+        return self.durable_client_input(
+            client_name, task_hub, connection_name, sync=True)
+
 
 class DFApp(Blueprint, FunctionRegister):
     """Durable Functions (DF) app.

azure-functions-durable/azure/durable_functions/internal/converters.py

@@ -180,8 +180,9 @@ def has_trigger_support(cls) -> bool:
 
     @classmethod
     def check_input_type_annotation(cls, pytype: type) -> bool:
-        from ..client import DurableFunctionsClient
-        return issubclass(pytype, (str, bytes, DurableFunctionsClient))
+        from ..client import DurableFunctionsClient, SyncDurableFunctionsClient
+        return issubclass(
+            pytype, (str, bytes, DurableFunctionsClient, SyncDurableFunctionsClient))
 
     @classmethod
     def check_output_type_annotation(cls, pytype: type) -> bool:
@@ -213,8 +214,10 @@ def encode(cls, obj: Any, *,
     @classmethod
     def decode(cls, data: meta.Datum, *,
                trigger_metadata: _TriggerMetadata) -> Any:
-        from ..client import DurableFunctionsClient
-        return DurableFunctionsClient(data.value)
+        # The decorator turns this host configuration into the requested sync or
+        # async rich client. Keeping conversion here as a string lets one binding
+        # type support both client shapes without a hidden sync bridge.
+        return data.value
 
 
 def register_durable_converters() -> None:

azure-functions-durable/azure/durable_functions/internal/history_export_compat.py

@@ -20,7 +20,6 @@
 
 from __future__ import annotations
 
-import atexit
 import threading
 from collections.abc import Mapping
 from datetime import datetime, timezone
@@ -217,30 +216,13 @@ def _close_sync_export_client() -> None:
             pass
 
 
-def _context_for(client: Any) -> HistoryExportContext:
-    """Return the per-process export context, building it once from *client*.
-
-    Uses the per-invocation injected client to build the synchronous client and
-    pairs it with the configured writer. The result is cached per process: the
-    endpoint and writer are stable for the app's lifetime, so the first
-    invocation in each process establishes the context for the rest. A lock
-    guards the first-build race so concurrent fan-out activities do not each open
-    a channel; the built client is closed at process exit.
-    """
-    global _export_context
-    if _export_context is not None:
-        return _export_context
-    with _context_lock:
-        if _export_context is None:
-            if _export_writer is None:
-                raise RuntimeError(
-                    "history export writer is not configured; pass a writer to "
-                    "DFApp.configure_history_export(writer=...) at app startup")
-            context = HistoryExportContext(
-                client=_build_sync_client(client), writer=_export_writer)
-            atexit.register(_close_sync_export_client)
-            _export_context = context
-    return _export_context
+def _context_for(client: TaskHubGrpcClient) -> HistoryExportContext:
+    """Resolve an export context for the invocation's native sync client."""
+    if _export_writer is None:
+        raise RuntimeError(
+            "history export writer is not configured; pass a writer to "
+            "DFApp.configure_history_export(writer=...) at app startup")
+    return HistoryExportContext(client=client, writer=_export_writer)
 
 
 def list_terminal_instances_client_bound(

durabletask/extensions/history_export/__init__.py

@@ -28,6 +28,8 @@
     run_list_terminal_instances,
 )
 from durabletask.extensions.history_export.client import (
+    AsyncExportHistoryClient,
+    AsyncExportHistoryJobClient,
     ExportHistoryClient,
     ExportHistoryJobClient,
 )
@@ -67,6 +69,8 @@
     "ExportFormatKind",
     "ExportHistoryClient",
     "ExportHistoryJobClient",
+    "AsyncExportHistoryClient",
+    "AsyncExportHistoryJobClient",
     "ExportJobConfiguration",
     "ExportJobCreationOptions",
     "ExportJobDescription",