Add sync and async client parity

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`. `DFApp.durable_client_input()` now injects
+the synchronous client into synchronous functions and the asynchronous client
+into coroutine functions. Both clients support scheduled-task and history-export
+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

@@ -2,7 +2,9 @@
 # Licensed under the MIT License.
 
 import asyncio
+import atexit
 import json
+import threading
 
 from datetime import datetime, timedelta
 from typing import Any, Optional, Union
@@ -14,10 +16,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
@@ -26,6 +32,10 @@
 from .internal.compat.purge_history_result import PurgeHistoryResult
 
 
+_sync_client_cache: dict[str, "SyncDurableFunctionsClient"] = {}
+_sync_client_cache_lock = threading.Lock()
+
+
 # Client class used for Durable Functions
 class DurableFunctionsClient(AsyncTaskHubGrpcClient):
     """A gRPC client passed to Durable Functions durable client bindings.
@@ -490,3 +500,121 @@ 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."""
+
+    taskHubName: str
+    connectionName: str
+    creationUrls: dict[str, str]
+    managementUrls: dict[str, str]
+    baseUrl: str
+    requiredQueryStringParameters: str
+    rpcBaseUrl: str
+    httpBaseUrl: str
+    maxGrpcMessageSizeInBytes: int
+    grpcHttpClientTimeout: timedelta | str
+
+    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)
+
+    @classmethod
+    def get_cached(cls, client_as_string: str) -> "SyncDurableFunctionsClient":
+        """Get the process-wide client for a durable-client binding configuration.
+
+        Synchronous Functions bindings can be invoked frequently by history
+        export fan-out activities. Reusing the gRPC channel avoids creating and
+        tearing down a channel for every activity invocation.
+        """
+        with _sync_client_cache_lock:
+            cached = _sync_client_cache.get(client_as_string)
+            if cached is None:
+                cached = cls(client_as_string)
+                _sync_client_cache[client_as_string] = cached
+            return cached
+
+    def _parse_client_configuration(self, client_as_string: str) -> None:
+        client = json.loads(client_as_string)
+        self.taskHubName = client.get("taskHubName") or ""
+        self.connectionName = client.get("connectionName") or ""
+        self.creationUrls = client.get("creationUrls") or {}
+        self.managementUrls = client.get("managementUrls") or {}
+        self.baseUrl = client.get("baseUrl") or ""
+        self.requiredQueryStringParameters = client.get(
+            "requiredQueryStringParameters") or ""
+        self.rpcBaseUrl = client.get("rpcBaseUrl") or ""
+        self.httpBaseUrl = client.get("httpBaseUrl") or ""
+        self.maxGrpcMessageSizeInBytes = client.get(
+            "maxGrpcMessageSizeInBytes") or 0
+        self.grpcHttpClientTimeout = client.get(
+            "grpcHttpClientTimeout") or timedelta(seconds=30)
+
+    def create_check_status_response(
+            self, request: func.HttpRequest, instance_id: str) -> func.HttpResponse:
+        payload = self._get_client_response_links(request, instance_id)
+        return func.HttpResponse(
+            body=str(payload),
+            status_code=202,
+            headers={
+                "content-type": "application/json",
+                "Location": payload["statusQueryGetUri"],
+            },
+        )
+
+    def create_http_management_payload(
+            self,
+            request: func.HttpRequest | str | None = None,
+            instance_id: str | None = None) -> HttpManagementPayload:
+        if instance_id is None and isinstance(request, str):
+            instance_id = request
+            request = None
+        if instance_id is None:
+            raise TypeError("instance_id is required")
+        resolved_request = request if isinstance(request, func.HttpRequest) else None
+        return self._get_client_response_links(resolved_request, instance_id)
+
+    def _get_client_response_links(
+            self, request: func.HttpRequest | None,
+            instance_id: str) -> HttpManagementPayload:
+        return HttpManagementPayload(
+            instance_id,
+            self._get_instance_status_url(request, instance_id),
+            self.requiredQueryStringParameters)
+
+    def _get_instance_status_url(
+            self, request: func.HttpRequest | None, instance_id: str) -> str:
+        encoded_instance_id = quote(instance_id)
+        if request is not None:
+            request_url = urlparse(request.url)
+            return (
+                f"{request_url.scheme}://{request_url.netloc}"
+                f"/runtime/webhooks/durabletask/instances/{encoded_instance_id}")
+        base_url = self.baseUrl.rstrip("/") if self.baseUrl else ""
+        return f"{base_url}/instances/{encoded_instance_id}"
+
+
+def _close_cached_sync_clients() -> None:
+    """Release process-wide synchronous durable-client channels on shutdown."""
+    with _sync_client_cache_lock:
+        clients = list(_sync_client_cache.values())
+        _sync_client_cache.clear()
+    for client in clients:
+        client.close()
+
+
+atexit.register(_close_cached_sync_clients)

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
@@ -356,10 +358,14 @@ 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,
                              ) -> Callable[[Callable[..., Any]], FunctionBuilder]:
         """Register a Durable-client Function.
 
+        Coroutine functions receive an asynchronous
+        :class:`DurableFunctionsClient`; synchronous functions receive a cached
+        :class:`SyncDurableFunctionsClient`.
+
         Parameters
         ----------
         client_name: str
@@ -379,11 +385,9 @@ def durable_client_input(self,
         @self._build_function
         def wrap(fb: FunctionBuilder) -> FunctionBuilder:
             def decorator() -> FunctionBuilder:
-                # The durableClient binding converter
-                # (``DurableClientConverter``) constructs the rich
-                # ``DurableFunctionsClient`` from the host-supplied configuration
-                # string during decode, so no per-function client middleware is
-                # needed here.
+                # The converter returns the host configuration string. The
+                # function wrapper below constructs the rich client matching
+                # the decorated function's synchronous or asynchronous type.
                 fb.add_binding(
                     binding=DurableClient(name=client_name,
                                           task_hub=task_hub,
@@ -393,16 +397,75 @@ def decorator() -> FunctionBuilder:
             return decorator()
 
         def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder:
-            # Expose the raw client function for unit testing, mirroring the
-            # ``.orchestrator_function`` and ``.entity_function`` attributes.
-            # Unlike v1 there is no client middleware wrapper (the durableClient
-            # binding converter builds the rich client during decode), so the
-            # 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)
+            # Expose the original client function for unit testing, mirroring
+            # ``.orchestrator_function`` and ``.entity_function``. The registered
+            # wrapper resolves the client from the binding configuration and
+            # closes it after the invocation.
+            from ..client import DurableFunctionsClient, SyncDurableFunctionsClient
+
+            function = (user_fn._function._func  # pyright: ignore[reportPrivateUsage]
+                        if isinstance(user_fn, FunctionBuilder) else user_fn)
+            signature = inspect.signature(function)
+            is_async_function = inspect.iscoroutinefunction(function)
+
+            def bind_client(
+                    args: tuple[Any, ...],
+                    kwargs: dict[str, Any],
+            ) -> tuple[inspect.BoundArguments, DurableFunctionsClient | SyncDurableFunctionsClient]:
+                bound = signature.bind(*args, **kwargs)
+                if client_name not in bound.arguments:
+                    raise TypeError(
+                        f"durable client binding parameter '{client_name}' is not "
+                        f"declared by function '{function.__name__}'")
+                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 = (DurableFunctionsClient(raw_client) if is_async_function
+                          else SyncDurableFunctionsClient.get_cached(raw_client))
+                bound.arguments[client_name] = client
+                return bound, client
+
+            def set_client_metadata(client_bound: Callable[..., Any]) -> None:
+                annotations = dict(function.__annotations__)
+                supported_client_types = (
+                    str,
+                    bytes,
+                    DurableFunctionsClient,
+                    SyncDurableFunctionsClient,
+                )
+                if annotations.get(client_name) not in supported_client_types:
+                    annotations[client_name] = str
+                client_bound.__annotations__ = annotations
+                setattr(client_bound, "client_function", function)
+
+            if is_async_function:
+                @wraps(function)
+                async def async_client_bound(*args: Any, **kwargs: Any) -> Any:
+                    bound, client = bind_client(args, kwargs)
+                    try:
+                        result = function(*bound.args, **bound.kwargs)
+                        return await result
+                    finally:
+                        if isinstance(client, DurableFunctionsClient):
+                            client.schedule_close()
+                client_bound = async_client_bound
+            else:
+                @wraps(function)
+                def sync_client_bound(*args: Any, **kwargs: Any) -> Any:
+                    bound, client = bind_client(args, kwargs)
+                    try:
+                        return function(*bound.args, **bound.kwargs)
+                    finally:
+                        if isinstance(client, DurableFunctionsClient):
+                            client.schedule_close()
+                client_bound = sync_client_bound
+
+            set_client_metadata(client_bound)
+            if isinstance(user_fn, FunctionBuilder):
+                user_fn._function._func = client_bound  # pyright: ignore[reportPrivateUsage]
+                return wrap(user_fn)
+            return wrap(client_bound)
 
         return attach_client_function
 

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,16 +214,19 @@ 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:
     """Register this package's durable binding converters with azure-functions.
 
     Overrides the SDK's built-in converters for the four durable binding types
-    so the host uses the durabletask-based encodings and the durable-client
-    binding is decoded to a :class:`DurableFunctionsClient`.
+    so the host uses the durabletask-based encodings. The durable-client binding
+    is decoded to its host configuration string, which the decorator converts to
+    the requested synchronous or asynchronous rich client per invocation.
     """
     func.register_converter(
         ORCHESTRATION_TRIGGER, OrchestrationTriggerConverter, overwrite=True)