From 231eff2abc3e19842793be9e8d818489bcf9e34b Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:09:25 -0600 Subject: [PATCH 01/14] Add sync and async client parity Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- CHANGELOG.md | 4 + azure-functions-durable/CHANGELOG.md | 9 + .../azure/durable_functions/__init__.py | 3 +- .../azure/durable_functions/client.py | 33 ++- .../decorators/durable_app.py | 57 ++++- .../durable_functions/internal/converters.py | 11 +- .../internal/history_export_compat.py | 32 +-- .../extensions/history_export/__init__.py | 4 + .../extensions/history_export/client.py | 201 +++++++++++++++++- durabletask/scheduled/__init__.py | 6 +- durabletask/scheduled/client.py | 116 +++++++++- .../test_converters.py | 37 ++-- .../test_decorator_compat.py | 19 +- .../test_history_export_compat.py | 27 +-- 14 files changed, 465 insertions(+), 94 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 03a7ff22..64cfb2ab 100644 --- a/CHANGELOG.md +++ b/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()` 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. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 4d1c5947..fbe9ca96 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/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 diff --git a/azure-functions-durable/azure/durable_functions/__init__.py b/azure-functions-durable/azure/durable_functions/__init__.py index a9fc6583..df5059a5 100644 --- a/azure-functions-durable/azure/durable_functions/__init__.py +++ b/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", diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index 479ab1cd..05ff83e7 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/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 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 9b2b247a..b86a2b78 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/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. diff --git a/azure-functions-durable/azure/durable_functions/internal/converters.py b/azure-functions-durable/azure/durable_functions/internal/converters.py index 09de202f..31f8813a 100644 --- a/azure-functions-durable/azure/durable_functions/internal/converters.py +++ b/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: diff --git a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py b/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py index b56ddcae..a2bc5a1c 100644 --- a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py +++ b/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( diff --git a/durabletask/extensions/history_export/__init__.py b/durabletask/extensions/history_export/__init__.py index 9cd191d1..ce84437f 100644 --- a/durabletask/extensions/history_export/__init__.py +++ b/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", diff --git a/durabletask/extensions/history_export/client.py b/durabletask/extensions/history_export/client.py index e9dbffe0..9a56bd10 100644 --- a/durabletask/extensions/history_export/client.py +++ b/durabletask/extensions/history_export/client.py @@ -52,9 +52,10 @@ from __future__ import annotations +import asyncio import time import uuid -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator from datetime import datetime, timezone from typing import Any, cast @@ -119,7 +120,12 @@ def _grpc_status(ex: grpc.RpcError) -> grpc.StatusCode | None: return result if isinstance(result, grpc.StatusCode) else None -__all__ = ["ExportHistoryClient", "ExportHistoryJobClient"] +__all__ = [ + "AsyncExportHistoryClient", + "AsyncExportHistoryJobClient", + "ExportHistoryClient", + "ExportHistoryJobClient", +] class ExportHistoryClient: @@ -473,3 +479,194 @@ def wait( def delete(self) -> None: """Delete the export job; see :meth:`ExportHistoryClient.delete_job`.""" self._parent.delete_job(self._job_id) + + +class AsyncExportHistoryClient: + """Asynchronous façade for creating and inspecting export jobs.""" + + def __init__( + self, + durable_task_client: client_module.AsyncTaskHubGrpcClient, + writer: HistoryWriter, + ) -> None: + self._client = durable_task_client + self._writer = writer + + async def create_job( + self, + options: ExportJobCreationOptions, + *, + job_id: str | None = None, + ) -> ExportJobDescription: + """Create a new export job and start its driving orchestrator.""" + config = options.to_configuration() + resolved_job_id = job_id or uuid.uuid4().hex + entity_id = entities.EntityInstanceId(ENTITY_NAME, resolved_job_id) + created_at = datetime.now(timezone.utc) + await self._client.signal_entity( + entity_id, + ExportJobEntity.OP_CREATE, + input={"config": config.to_dict(), "created_at": created_at.isoformat()}, + ) + logger.info( + "Submitted export job %r; orchestrator instance ID will be %s", + resolved_job_id, orchestrator_instance_id_for(resolved_job_id), + ) + return ExportJobDescription( + job_id=resolved_job_id, + status=ExportJobStatus.ACTIVE, + created_at=created_at, + last_modified_at=created_at, + config=config, + orchestrator_instance_id=orchestrator_instance_id_for(resolved_job_id), + scanned_instances=0, + exported_instances=0, + failed_instances=0, + last_error=None, + checkpoint=None, + last_checkpoint_time=None, + ) + + async def get_job(self, job_id: str) -> ExportJobDescription | None: + """Look up an export job by ID, or return None if it does not exist.""" + entity_id = entities.EntityInstanceId(ENTITY_NAME, job_id) + meta = await self._client.get_entity(entity_id, include_state=True) + if meta is None: + return None + state = meta.get_typed_state() + if not isinstance(state, dict): + return None + return ExportJobDescription.from_state_dict(job_id, cast("dict[str, Any]", state)) + + async def list_jobs( + self, query: ExportJobQuery | None = None) -> AsyncIterator[ExportJobDescription]: + """Asynchronously enumerate export jobs matching *query*.""" + if query is None: + query = ExportJobQuery() + entity_query = client_module.EntityQuery( + instance_id_starts_with=_ENTITY_ID_PREFIX, + last_modified_from=query.last_modified_from, + last_modified_to=query.last_modified_to, + include_state=True, + page_size=query.page_size, + ) + status_filter = set(query.status) if query.status else None + for meta in await self._client.get_all_entities(entity_query): + if meta.id.entity != ENTITY_NAME.lower(): + continue + state = meta.get_typed_state() + if not isinstance(state, dict): + continue + try: + desc = ExportJobDescription.from_state_dict( + meta.id.key, cast("dict[str, Any]", state)) + except (KeyError, ValueError) as ex: + logger.warning( + "list_jobs: skipping export-job entity %r; state did not match " + "the current schema (%s)", meta.id.key, ex) + continue + if status_filter is None or desc.status in status_filter: + yield desc + + async def wait_for_job( + self, + job_id: str, + *, + timeout: float = 300.0, + poll_interval: float = 1.0, + ) -> ExportJobDescription: + """Wait asynchronously until a job is terminal.""" + if timeout <= 0: + raise ValueError("timeout must be positive") + if poll_interval <= 0: + raise ValueError("poll_interval must be positive") + deadline = time.monotonic() + timeout + last: ExportJobDescription | None = None + while True: + desc = await self.get_job(job_id) + if desc is not None: + last = desc + if desc.status in _TERMINAL_STATUSES: + return desc + if time.monotonic() >= deadline: + if last is None: + raise ExportJobNotFoundError(job_id) + raise TimeoutError( + f"Export job '{job_id}' did not reach a terminal status " + f"within {timeout}s (last status: {last.status.value})") + await asyncio.sleep(poll_interval) + + async def delete_job(self, job_id: str) -> None: + """Stop and delete an export job.""" + entity_id = entities.EntityInstanceId(ENTITY_NAME, job_id) + orch_instance_id = orchestrator_instance_id_for(job_id) + await self._client.signal_entity(entity_id, ExportJobEntity.OP_DELETE) + try: + await self._client.terminate_orchestration(orch_instance_id, recursive=True) + except grpc.RpcError as ex: + if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND: + raise + try: + await self._client.wait_for_orchestration_completion( + orch_instance_id, timeout=_DELETE_WAIT_TIMEOUT_SECONDS) + except TimeoutError: + logger.warning( + "Export job %r orchestrator %r did not terminate within %ss; " + "continuing with purge anyway", + job_id, orch_instance_id, _DELETE_WAIT_TIMEOUT_SECONDS) + except grpc.RpcError as ex: + if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND: + raise + try: + await self._client.purge_orchestration(orch_instance_id, recursive=True) + except grpc.RpcError as ex: + if _grpc_status(ex) != grpc.StatusCode.NOT_FOUND: + raise + + async def cancel_job(self, job_id: str) -> None: + """Alias for :meth:`delete_job`.""" + await self.delete_job(job_id) + + def get_job_client(self, job_id: str) -> "AsyncExportHistoryJobClient": + """Return an asynchronous per-job façade for *job_id*.""" + return AsyncExportHistoryJobClient(self, job_id) + + @property + def writer(self) -> HistoryWriter: + """Get the history writer used by this client.""" + return self._writer + + +class AsyncExportHistoryJobClient: + """Asynchronous per-job façade returned by AsyncExportHistoryClient.""" + + def __init__(self, parent: AsyncExportHistoryClient, job_id: str) -> None: + if not job_id: + raise ValueError("job_id must be a non-empty string") + self._parent = parent + self._job_id = job_id + + @property + def job_id(self) -> str: + """Get this export job's ID.""" + return self._job_id + + @property + def orchestrator_instance_id(self) -> str: + """Get the driving orchestrator's ID.""" + return orchestrator_instance_id_for(self._job_id) + + async def describe(self) -> ExportJobDescription | None: + """Fetch the latest description, or None if the job is missing.""" + return await self._parent.get_job(self._job_id) + + async def wait( + self, *, timeout: float = 300.0, poll_interval: float = 1.0, + ) -> ExportJobDescription: + """Wait asynchronously until this job is terminal.""" + return await self._parent.wait_for_job( + self._job_id, timeout=timeout, poll_interval=poll_interval) + + async def delete(self) -> None: + """Delete this export job.""" + await self._parent.delete_job(self._job_id) diff --git a/durabletask/scheduled/__init__.py b/durabletask/scheduled/__init__.py index 57d295ba..39e95f99 100644 --- a/durabletask/scheduled/__init__.py +++ b/durabletask/scheduled/__init__.py @@ -9,7 +9,9 @@ manage schedules from the client via :class:`ScheduledTaskClient`. """ -from durabletask.scheduled.client import ScheduleClient, ScheduledTaskClient +from durabletask.scheduled.client import (AsyncScheduleClient, + AsyncScheduledTaskClient, + ScheduleClient, ScheduledTaskClient) from durabletask.scheduled.exceptions import (ScheduleClientValidationError, ScheduleError, ScheduleInvalidTransitionError, @@ -21,7 +23,9 @@ __all__ = [ "ScheduledTaskClient", + "AsyncScheduledTaskClient", "ScheduleClient", + "AsyncScheduleClient", "ScheduleCreationOptions", "ScheduleUpdateOptions", "ScheduleDescription", diff --git a/durabletask/scheduled/client.py b/durabletask/scheduled/client.py index dc6767ac..c7eb47d0 100644 --- a/durabletask/scheduled/client.py +++ b/durabletask/scheduled/client.py @@ -3,8 +3,8 @@ import logging -from durabletask.client import (EntityQuery, OrchestrationStatus, - TaskHubGrpcClient) +from durabletask.client import (AsyncTaskHubGrpcClient, EntityQuery, + OrchestrationStatus, TaskHubGrpcClient) from durabletask.entities import EntityInstanceId from durabletask.internal.helpers import ensure_aware from durabletask.scheduled import transitions @@ -152,3 +152,115 @@ def _matches_filter(state: ScheduleState, schedule_query: ScheduleQuery | None) if schedule_query.created_to is not None and not (created_at and created_at < schedule_query.created_to): return False return True + + +class AsyncScheduleClient: + """Asynchronous client for managing a single schedule instance.""" + + def __init__(self, client: AsyncTaskHubGrpcClient, schedule_id: str, + *, operation_timeout: float = 60): + if not schedule_id: + raise ValueError("schedule_id cannot be empty.") + self._client = client + self._schedule_id = schedule_id + self._entity_id = EntityInstanceId(ENTITY_NAME, schedule_id) + self._operation_timeout = operation_timeout + + @property + def schedule_id(self) -> str: + """Gets the ID of this schedule.""" + return self._schedule_id + + async def _run_operation( + self, operation_name: str, input: object | None = None) -> None: + request = ScheduleOperationRequest( + entity_id=str(self._entity_id), + operation_name=operation_name, + input=input, + ) + instance_id = await self._client.schedule_new_orchestration( + execute_schedule_operation_orchestrator, input=request) + state = await self._client.wait_for_orchestration_completion( + instance_id, timeout=self._operation_timeout) + if state is None or state.runtime_status != OrchestrationStatus.COMPLETED: + failure = state.failure_details if state else None + message = failure.message if failure else "unknown error" + raise RuntimeError( + f"Failed to '{operation_name}' schedule '{self._schedule_id}': {message}") + + async def create(self, options: ScheduleCreationOptions) -> None: + """Create or update this schedule with the given configuration.""" + await self._run_operation(transitions.CREATE_SCHEDULE, options) + + async def update(self, options: ScheduleUpdateOptions) -> None: + """Update this schedule's configuration.""" + await self._run_operation(transitions.UPDATE_SCHEDULE, options) + + async def pause(self) -> None: + """Pause this schedule.""" + await self._run_operation(transitions.PAUSE_SCHEDULE) + + async def resume(self) -> None: + """Resume this schedule.""" + await self._run_operation(transitions.RESUME_SCHEDULE) + + async def delete(self) -> None: + """Delete this schedule.""" + await self._run_operation(DELETE_OPERATION) + + async def describe(self) -> ScheduleDescription: + """Retrieve the current details of this schedule.""" + metadata = await self._client.get_entity(self._entity_id, include_state=True) + if metadata is None: + raise ScheduleNotFoundError(self._schedule_id) + state = metadata.get_typed_state(ScheduleState) + if state is None: + raise ScheduleNotFoundError(self._schedule_id) + return state.to_description() + + +class AsyncScheduledTaskClient: + """Asynchronous client for managing scheduled tasks.""" + + def __init__(self, client: AsyncTaskHubGrpcClient, *, + operation_timeout: float = 60): + self._client = client + self._operation_timeout = operation_timeout + + def get_schedule_client(self, schedule_id: str) -> AsyncScheduleClient: + """Get an asynchronous handle for a specific schedule.""" + return AsyncScheduleClient( + self._client, schedule_id, operation_timeout=self._operation_timeout) + + async def create_schedule( + self, options: ScheduleCreationOptions) -> AsyncScheduleClient: + """Create a new schedule and return a client for managing it.""" + schedule_client = self.get_schedule_client(options.schedule_id) + await schedule_client.create(options) + return schedule_client + + async def get_schedule(self, schedule_id: str) -> ScheduleDescription | None: + """Get a schedule description by ID, or None if it does not exist.""" + try: + return await self.get_schedule_client(schedule_id).describe() + except ScheduleNotFoundError: + return None + + async def list_schedules( + self, schedule_query: ScheduleQuery | None = None) -> list[ScheduleDescription]: + """List schedules matching the given filter criteria.""" + prefix = schedule_query.schedule_id_prefix if schedule_query and schedule_query.schedule_id_prefix else "" + page_size = (schedule_query.page_size if schedule_query and schedule_query.page_size + else ScheduleQuery.DEFAULT_PAGE_SIZE) + query = EntityQuery( + instance_id_starts_with=f"@{ENTITY_NAME}@{prefix}", + include_state=True, + page_size=page_size, + ) + results: list[ScheduleDescription] = [] + for metadata in await self._client.get_all_entities(query): + state = metadata.get_typed_state(ScheduleState) + if state is not None and state.schedule_configuration is not None and \ + ScheduledTaskClient._matches_filter(state, schedule_query): + results.append(state.to_description()) + return results diff --git a/tests/azure-functions-durable/test_converters.py b/tests/azure-functions-durable/test_converters.py index 36308bda..69ff1b8d 100644 --- a/tests/azure-functions-durable/test_converters.py +++ b/tests/azure-functions-durable/test_converters.py @@ -9,7 +9,6 @@ that the converters use the durabletask-based encodings the host expects. """ -from unittest.mock import patch import json @@ -110,8 +109,13 @@ def test_activity_trigger_decode_falls_back_to_raw_string(): # --------------------------------------------------------------------------- def test_durable_client_accepts_client_and_string_annotations(): - from azure.durable_functions.client import DurableFunctionsClient + from azure.durable_functions.client import ( + DurableFunctionsClient, + SyncDurableFunctionsClient, + ) assert DurableClientConverter.check_input_type_annotation(DurableFunctionsClient) + assert DurableClientConverter.check_input_type_annotation( + SyncDurableFunctionsClient) assert DurableClientConverter.check_input_type_annotation(str) assert DurableClientConverter.check_input_type_annotation(bytes) assert not DurableClientConverter.check_input_type_annotation(int) @@ -122,23 +126,14 @@ def test_durable_client_has_no_trigger_support_or_implicit_output(): assert DurableClientConverter.has_implicit_output() is False -def test_durable_client_decode_constructs_client_from_value(): - with patch( - "azure.durable_functions.client.DurableFunctionsClient" - ) as mock_client: - result = DurableClientConverter.decode( - meta.Datum(type="string", value="client-config"), - trigger_metadata=None) - mock_client.assert_called_once_with("client-config") - assert result is mock_client.return_value +def test_durable_client_decode_returns_host_configuration(): + result = DurableClientConverter.decode( + meta.Datum(type="string", value="client-config"), + trigger_metadata=None) + assert result == "client-config" -async def test_durable_client_decode_builds_working_client_from_config(): - # End-to-end: decode parses a host-style configuration string into a live - # DurableFunctionsClient (the construction that previously lived in the - # decorator's client middleware). - from azure.durable_functions.client import DurableFunctionsClient - +def test_durable_client_decode_preserves_host_configuration(): config = json.dumps({ "taskHubName": "TestHub", "requiredQueryStringParameters": "code=xyz", @@ -146,10 +141,6 @@ async def test_durable_client_decode_builds_working_client_from_config(): "rpcBaseUrl": "http://localhost:8080/", }) - client = DurableClientConverter.decode( + value = DurableClientConverter.decode( meta.Datum(type="string", value=config), trigger_metadata=None) - assert isinstance(client, DurableFunctionsClient) - try: - assert client.taskHubName == "TestHub" - finally: - await client.close() + assert value == config diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index 242ae33c..fcc94842 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -123,17 +123,24 @@ async def starter(client): assert binding.connection_name == "conn" -async def test_durable_client_input_does_not_wrap_user_function(): - # The durableClient binding converter (``DurableClientConverter``) - # constructs the rich ``DurableFunctionsClient`` during decode, so the - # decorator no longer wraps the user function in client-building middleware. +async def test_durable_client_input_wraps_host_configuration_as_async_client(): app = df.DFApp() async def starter(client): - return None + return type(client).__name__ fb = app.durable_client_input(client_name="client")(starter) - assert fb._function._func is starter + assert await fb._function._func(client="{}") == "DurableFunctionsClient" + + +async def test_durable_client_input_sync_injects_sync_client(): + app = df.DFApp() + + def starter(client): + return type(client).__name__ + + fb = app.durable_client_input_sync(client_name="client")(starter) + assert await fb._function._func(client="{}") == "SyncDurableFunctionsClient" # --------------------------------------------------------------------------- diff --git a/tests/azure-functions-durable/test_history_export_compat.py b/tests/azure-functions-durable/test_history_export_compat.py index 74521cfa..d82076d8 100644 --- a/tests/azure-functions-durable/test_history_export_compat.py +++ b/tests/azure-functions-durable/test_history_export_compat.py @@ -189,36 +189,17 @@ def test_continuous_create_on_fresh_entity_is_marked_failed(): # is built once from the injected client and closed at interpreter exit. # --------------------------------------------------------------------------- -def test_context_for_builds_caches_and_closes_sync_client(monkeypatch): +def test_context_for_uses_invocation_sync_client(monkeypatch): fake_client = MagicMock() - monkeypatch.setattr(hec, "_build_sync_client", lambda _c: fake_client) writer = MagicMock() monkeypatch.setattr(hec, "_export_writer", writer) - monkeypatch.setattr(hec, "_export_context", None) - - registered = [] - monkeypatch.setattr(hec.atexit, "register", lambda fn: registered.append(fn)) - context = hec._context_for(object()) + context = hec._context_for(fake_client) - # The context pairs the built sync client with the configured writer and is - # cached for the process. + # The context pairs the native injected client with the configured writer. assert context.client is fake_client assert context.writer is writer - assert hec._export_context is context - assert hec._close_sync_export_client in registered - - # A second resolution is a no-op: same cached context, no duplicate - # registration and no new client build. - assert hec._context_for(object()) is context - assert registered.count(hec._close_sync_export_client) == 1 - - # Closing releases the client, resets state, and is idempotent + safe. - hec._close_sync_export_client() - fake_client.close.assert_called_once() - assert hec._export_context is None - hec._close_sync_export_client() - fake_client.close.assert_called_once() + assert hec._context_for(fake_client) is not context def test_context_for_requires_configured_writer(monkeypatch): From 1ed37e9bbdd16c8e7dd31ea9357b00c5ab95e9e4 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:15:45 -0600 Subject: [PATCH 02/14] Expose schedule filter helper Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- durabletask/scheduled/client.py | 6 +++--- .../scheduled/test_client_filter_and_capability.py | 14 +++++++------- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/durabletask/scheduled/client.py b/durabletask/scheduled/client.py index c7eb47d0..5d3f4572 100644 --- a/durabletask/scheduled/client.py +++ b/durabletask/scheduled/client.py @@ -131,13 +131,13 @@ def list_schedules(self, schedule_query: ScheduleQuery | None = None) -> list[Sc state = metadata.get_typed_state(ScheduleState) if state is None or state.schedule_configuration is None: continue - if not self._matches_filter(state, schedule_query): + if not self.matches_filter(state, schedule_query): continue results.append(state.to_description()) return results @staticmethod - def _matches_filter(state: ScheduleState, schedule_query: ScheduleQuery | None) -> bool: + def matches_filter(state: ScheduleState, schedule_query: ScheduleQuery | None) -> bool: if schedule_query is None: return True if schedule_query.status is not None and state.status != schedule_query.status: @@ -261,6 +261,6 @@ async def list_schedules( for metadata in await self._client.get_all_entities(query): state = metadata.get_typed_state(ScheduleState) if state is not None and state.schedule_configuration is not None and \ - ScheduledTaskClient._matches_filter(state, schedule_query): + ScheduledTaskClient.matches_filter(state, schedule_query): results.append(state.to_description()) return results diff --git a/tests/durabletask/scheduled/test_client_filter_and_capability.py b/tests/durabletask/scheduled/test_client_filter_and_capability.py index 66cfc318..5533ce15 100644 --- a/tests/durabletask/scheduled/test_client_filter_and_capability.py +++ b/tests/durabletask/scheduled/test_client_filter_and_capability.py @@ -30,7 +30,7 @@ def test_naive_bound_does_not_crash_against_aware_created_at(self): # A naive bound is normalized by ScheduleQuery, so the comparison must # not raise "can't compare offset-naive and offset-aware". q = ScheduleQuery(created_from=datetime(2026, 1, 1, 0, 0, 0)) - assert ScheduledTaskClient._matches_filter(state, q) is True + assert ScheduledTaskClient.matches_filter(state, q) is True def test_created_from_is_exclusive(self): # Bounds match .NET's exclusive semantics: a schedule created exactly at @@ -38,13 +38,13 @@ def test_created_from_is_exclusive(self): boundary = datetime(2026, 1, 1, tzinfo=timezone.utc) state = _state_created_at(boundary) q = ScheduleQuery(created_from=boundary) - assert ScheduledTaskClient._matches_filter(state, q) is False + assert ScheduledTaskClient.matches_filter(state, q) is False def test_created_to_is_exclusive(self): boundary = datetime(2026, 1, 1, tzinfo=timezone.utc) state = _state_created_at(boundary) q = ScheduleQuery(created_to=boundary) - assert ScheduledTaskClient._matches_filter(state, q) is False + assert ScheduledTaskClient.matches_filter(state, q) is False def test_inside_window_matches(self): state = _state_created_at(datetime(2026, 1, 15, tzinfo=timezone.utc)) @@ -52,17 +52,17 @@ def test_inside_window_matches(self): created_from=datetime(2026, 1, 1, tzinfo=timezone.utc), created_to=datetime(2026, 2, 1, tzinfo=timezone.utc), ) - assert ScheduledTaskClient._matches_filter(state, q) is True + assert ScheduledTaskClient.matches_filter(state, q) is True def test_outside_window_is_excluded(self): state = _state_created_at(datetime(2026, 3, 1, tzinfo=timezone.utc)) q = ScheduleQuery(created_to=datetime(2026, 2, 1, tzinfo=timezone.utc)) - assert ScheduledTaskClient._matches_filter(state, q) is False + assert ScheduledTaskClient.matches_filter(state, q) is False def test_status_filter(self): state = _state_created_at(datetime(2026, 1, 1, tzinfo=timezone.utc)) - assert ScheduledTaskClient._matches_filter(state, ScheduleQuery(status=ScheduleStatus.ACTIVE)) is True - assert ScheduledTaskClient._matches_filter(state, ScheduleQuery(status=ScheduleStatus.PAUSED)) is False + assert ScheduledTaskClient.matches_filter(state, ScheduleQuery(status=ScheduleStatus.ACTIVE)) is True + assert ScheduledTaskClient.matches_filter(state, ScheduleQuery(status=ScheduleStatus.PAUSED)) is False class TestScheduledTasksCapability: From 4cce8a88986fbe387533c296f9fe8ddd0c4f871a Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:16:40 -0600 Subject: [PATCH 03/14] Fix Functions client typing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../azure/durable_functions/client.py | 74 +++++++++++++++++-- .../decorators/durable_app.py | 14 ++-- .../internal/history_export_compat.py | 52 ------------- .../test_history_export_compat.py | 13 ---- 4 files changed, 76 insertions(+), 77 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index 05ff83e7..1574caa4 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -499,6 +499,17 @@ def _create_http_response(status_code: int, body: Union[str, Any]) -> func.HttpR 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( @@ -516,8 +527,61 @@ def __init__(self, client_as_string: str): 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 + 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}" 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 b86a2b78..f28308dd 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -409,8 +409,8 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder: # ``FunctionBuilder`` (e.g. history export), which is left untouched. from ..client import DurableFunctionsClient, SyncDurableFunctionsClient - function = (user_fn._function._func if isinstance(user_fn, FunctionBuilder) - else user_fn) + function = (user_fn._function._func # pyright: ignore[reportPrivateUsage] + if isinstance(user_fn, FunctionBuilder) else user_fn) signature = inspect.signature(function) @wraps(function) @@ -427,15 +427,15 @@ async def client_bound(*args: Any, **kwargs: Any) -> Any: result = function(*bound.args, **bound.kwargs) return await result if inspect.isawaitable(result) else result finally: - if sync: - client.close() - else: + if isinstance(client, DurableFunctionsClient): client.schedule_close() + else: + client.close() client_bound.__annotations__[client_name] = str - client_bound.client_function = function # pyright: ignore[reportFunctionMemberAccess] + setattr(client_bound, "client_function", function) if isinstance(user_fn, FunctionBuilder): - user_fn._function._func = client_bound + user_fn._function._func = client_bound # pyright: ignore[reportPrivateUsage] return wrap(user_fn) return wrap(client_bound) diff --git a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py b/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py index a2bc5a1c..2bf7d0bb 100644 --- a/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py +++ b/azure-functions-durable/azure/durable_functions/internal/history_export_compat.py @@ -20,7 +20,6 @@ from __future__ import annotations -import threading from collections.abc import Mapping from datetime import datetime, timezone from typing import Any, Optional, cast @@ -48,10 +47,6 @@ from durabletask.extensions.history_export.transitions import assert_valid_transition from durabletask.extensions.history_export.writer import HistoryWriter -from .azurefunctions_grpc_interceptor import ( - AzureFunctionsDefaultClientInterceptorImpl, -) -from .serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER # The activity registers under the same name the export orchestrator calls, so # it transparently replaces the core activity. @@ -156,11 +151,6 @@ def list_terminal_instances( # --------------------------------------------------------------------------- _export_writer: Optional[HistoryWriter] = None -# The per-process export context (sync client + writer), built lazily from the -# first injected durable client and reused across every export activity. Guarded -# by ``_context_lock`` for the first-build race; its client is closed at exit. -_export_context: Optional[HistoryExportContext] = None -_context_lock = threading.Lock() def set_export_writer(writer: HistoryWriter) -> None: @@ -174,48 +164,6 @@ def set_export_writer(writer: HistoryWriter) -> None: _export_writer = writer -def _build_sync_client(client: Any) -> TaskHubGrpcClient: - """Build a synchronous ``TaskHubGrpcClient`` from an injected durable client. - - The ``durable_client_input`` binding yields an async ``DurableFunctionsClient`` - carrying the host's RPC endpoint and auth; the export activities use the - synchronous client, so this bridges to one aimed at the same endpoint. - - > [!NOTE] - > This async->sync adapter is temporary. Once a first-class synchronous - > durable-client binding exists - > (https://github.com/microsoft/durabletask-python/issues/181), the export - > activities can be injected with a sync client directly and this bridge -- - > along with the separate channel it opens -- can be removed. - """ - interceptors = [AzureFunctionsDefaultClientInterceptorImpl( - client.taskHubName, client.requiredQueryStringParameters)] - return TaskHubGrpcClient( - host_address=client.rpcBaseUrl, - secure_channel=False, - interceptors=interceptors, - data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) - - -def _close_sync_export_client() -> None: - """Close the process-wide sync export client (registered via ``atexit``). - - The client is built once per worker process and reused across every export - activity, so it lives for the app's lifetime; closing it at interpreter - exit releases its gRPC channel on graceful shutdown. Idempotent and - exception-safe -- shutdown must never surface an error from cleanup. - """ - global _export_context - with _context_lock: - context = _export_context - _export_context = None - if context is not None: - try: - context.client.close() - except Exception: - pass - - def _context_for(client: TaskHubGrpcClient) -> HistoryExportContext: """Resolve an export context for the invocation's native sync client.""" if _export_writer is None: diff --git a/tests/azure-functions-durable/test_history_export_compat.py b/tests/azure-functions-durable/test_history_export_compat.py index d82076d8..e1a67c6d 100644 --- a/tests/azure-functions-durable/test_history_export_compat.py +++ b/tests/azure-functions-durable/test_history_export_compat.py @@ -203,20 +203,7 @@ def test_context_for_uses_invocation_sync_client(monkeypatch): def test_context_for_requires_configured_writer(monkeypatch): - monkeypatch.setattr(hec, "_build_sync_client", lambda _c: MagicMock()) monkeypatch.setattr(hec, "_export_writer", None) - monkeypatch.setattr(hec, "_export_context", None) with pytest.raises(RuntimeError, match="writer is not configured"): hec._context_for(object()) - - -def test_close_sync_export_client_swallows_errors(monkeypatch): - fake_client = MagicMock() - fake_client.close.side_effect = RuntimeError("boom") - context = hec.HistoryExportContext(client=fake_client, writer=MagicMock()) - monkeypatch.setattr(hec, "_export_context", context) - - # Cleanup at shutdown must never raise. - hec._close_sync_export_client() - assert hec._export_context is None From 12bd0854f6bf076899fe957a53da8091741752d2 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:31:00 -0600 Subject: [PATCH 04/14] Add async client parity coverage Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../decorators/durable_app.py | 19 ++++------- .../durable_functions/internal/converters.py | 5 +-- durabletask/scheduled/__init__.py | 3 +- .../extensions/history_export/test_client.py | 27 ++++++++++++++++ .../scheduled/test_scheduled_e2e.py | 32 +++++++++++++++++-- 5 files changed, 68 insertions(+), 18 deletions(-) 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 f28308dd..201f3545 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -386,11 +386,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 and closes the requested + # synchronous or asynchronous rich client per invocation. fb.add_binding( binding=DurableClient(name=client_name, task_hub=task_hub, @@ -400,13 +398,10 @@ 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. + # 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] diff --git a/azure-functions-durable/azure/durable_functions/internal/converters.py b/azure-functions-durable/azure/durable_functions/internal/converters.py index 31f8813a..c0892f4c 100644 --- a/azure-functions-durable/azure/durable_functions/internal/converters.py +++ b/azure-functions-durable/azure/durable_functions/internal/converters.py @@ -224,8 +224,9 @@ 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) diff --git a/durabletask/scheduled/__init__.py b/durabletask/scheduled/__init__.py index 39e95f99..bfdb1549 100644 --- a/durabletask/scheduled/__init__.py +++ b/durabletask/scheduled/__init__.py @@ -6,7 +6,8 @@ This package provides a recurring schedule feature built on top of durable entities and a helper orchestrator. Enable it on a worker via :meth:`durabletask.worker.TaskHubGrpcWorker.configure_scheduled_tasks`, then -manage schedules from the client via :class:`ScheduledTaskClient`. +manage schedules from a synchronous :class:`ScheduledTaskClient` or +asynchronous :class:`AsyncScheduledTaskClient`. """ from durabletask.scheduled.client import (AsyncScheduleClient, diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index 3275b50f..e853ab48 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -19,6 +19,7 @@ from durabletask import client, task, worker from durabletask.extensions.history_export import ( + AsyncExportHistoryClient, ExportDestination, ExportFormat, ExportFormatKind, @@ -148,6 +149,32 @@ def test_create_get_and_wait_for_job_end_to_end( assert first["kind"] == "metadata" +async def test_async_client_create_list_wait_and_delete( + writer, seeded_terminal_instances, +): + async with client.AsyncTaskHubGrpcClient(host_address=HOST) as dt_client: + export_client = AsyncExportHistoryClient(dt_client, writer) + now = datetime.now(timezone.utc) + desc = await export_client.create_job(ExportJobCreationOptions( + mode=ExportMode.BATCH, + completed_time_from=now - timedelta(hours=1), + completed_time_to=now + timedelta(hours=1), + destination=ExportDestination(container="exports", prefix="async"), + format=ExportFormat(kind=ExportFormatKind.JSON), + )) + + final = await export_client.wait_for_job( + desc.job_id, timeout=30, poll_interval=0.1) + assert final.status == ExportJobStatus.COMPLETED + assert (await export_client.get_job(desc.job_id)) is not None + assert desc.job_id in { + job.job_id async for job in export_client.list_jobs() + } + + await export_client.delete_job(desc.job_id) + assert await export_client.get_job(desc.job_id) is None + + def test_get_job_returns_none_for_unknown_id(export_client): assert export_client.get_job("does-not-exist") is None diff --git a/tests/durabletask/scheduled/test_scheduled_e2e.py b/tests/durabletask/scheduled/test_scheduled_e2e.py index 4a158333..f0397ad4 100644 --- a/tests/durabletask/scheduled/test_scheduled_e2e.py +++ b/tests/durabletask/scheduled/test_scheduled_e2e.py @@ -10,9 +10,10 @@ import pytest from durabletask import client, task, worker -from durabletask.scheduled import (ScheduledTaskClient, ScheduleCreationOptions, - ScheduleQuery, ScheduleStatus, - ScheduleUpdateOptions) +from durabletask.scheduled import (AsyncScheduledTaskClient, + ScheduledTaskClient, + ScheduleCreationOptions, ScheduleQuery, + ScheduleStatus, ScheduleUpdateOptions) from durabletask.testing import create_test_backend from tests.durabletask._port_utils import find_free_port @@ -113,6 +114,31 @@ def test_get_nonexistent_returns_none(): assert scheduled.get_schedule("does-not-exist") is None +async def test_async_client_create_list_and_delete(): + with _make_worker() as w: + w.start() + async with client.AsyncTaskHubGrpcClient(host_address=HOST) as c: + scheduled = AsyncScheduledTaskClient(c) + schedule = await scheduled.create_schedule(ScheduleCreationOptions( + schedule_id="async-schedule", + orchestration_name=task.get_name(target_orchestrator), + interval=timedelta(seconds=30), + start_at=datetime.now(timezone.utc) + timedelta(hours=1), + )) + + assert (await scheduled.get_schedule("async-schedule")) is not None + listed = await scheduled.list_schedules( + ScheduleQuery(schedule_id_prefix="async-")) + assert [item.schedule_id for item in listed] == ["async-schedule"] + + await schedule.pause() + assert (await schedule.describe()).status == ScheduleStatus.PAUSED + await schedule.resume() + assert (await schedule.describe()).status == ScheduleStatus.ACTIVE + await schedule.delete() + assert await scheduled.get_schedule("async-schedule") is None + + def test_pause_and_resume(): with _make_worker() as w: w.start() From 49eaa349c7db807edc9634c93ff3d86b1d10d060 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:32:15 -0600 Subject: [PATCH 05/14] Cache sync Functions clients Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../azure/durable_functions/client.py | 33 +++++++++++++++++++ .../decorators/durable_app.py | 4 +-- .../test_decorator_compat.py | 4 +++ 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index 1574caa4..537c0ed8 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/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 @@ -30,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. @@ -527,6 +533,21 @@ def __init__(self, client_as_string: str): 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 "" @@ -585,3 +606,15 @@ def _get_instance_status_url( 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) 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 201f3545..27ca73bc 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -415,7 +415,7 @@ async def client_bound(*args: Any, **kwargs: Any) -> Any: 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 + client = (SyncDurableFunctionsClient.get_cached(raw_client) if sync else DurableFunctionsClient(raw_client)) bound.arguments[client_name] = client try: @@ -424,8 +424,6 @@ async def client_bound(*args: Any, **kwargs: Any) -> Any: finally: if isinstance(client, DurableFunctionsClient): client.schedule_close() - else: - client.close() client_bound.__annotations__[client_name] = str setattr(client_bound, "client_function", function) diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index fcc94842..b74c9ede 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -135,12 +135,16 @@ async def starter(client): async def test_durable_client_input_sync_injects_sync_client(): app = df.DFApp() + clients = [] def starter(client): + clients.append(client) return type(client).__name__ fb = app.durable_client_input_sync(client_name="client")(starter) assert await fb._function._func(client="{}") == "SyncDurableFunctionsClient" + assert await fb._function._func(client="{}") == "SyncDurableFunctionsClient" + assert clients[0] is clients[1] # --------------------------------------------------------------------------- From b8648bea0b98b64a5500c7f87a481a914dfffe85 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:48:33 -0600 Subject: [PATCH 06/14] Format async client signatures Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../extensions/history_export/client.py | 4 +++- durabletask/scheduled/client.py | 19 ++++++++++++++----- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/durabletask/extensions/history_export/client.py b/durabletask/extensions/history_export/client.py index 9a56bd10..f1e5392f 100644 --- a/durabletask/extensions/history_export/client.py +++ b/durabletask/extensions/history_export/client.py @@ -539,7 +539,9 @@ async def get_job(self, job_id: str) -> ExportJobDescription | None: return ExportJobDescription.from_state_dict(job_id, cast("dict[str, Any]", state)) async def list_jobs( - self, query: ExportJobQuery | None = None) -> AsyncIterator[ExportJobDescription]: + self, + query: ExportJobQuery | None = None, + ) -> AsyncIterator[ExportJobDescription]: """Asynchronously enumerate export jobs matching *query*.""" if query is None: query = ExportJobQuery() diff --git a/durabletask/scheduled/client.py b/durabletask/scheduled/client.py index 5d3f4572..819dcb0a 100644 --- a/durabletask/scheduled/client.py +++ b/durabletask/scheduled/client.py @@ -172,7 +172,9 @@ def schedule_id(self) -> str: return self._schedule_id async def _run_operation( - self, operation_name: str, input: object | None = None) -> None: + self, + operation_name: str, + input: object | None = None) -> None: request = ScheduleOperationRequest( entity_id=str(self._entity_id), operation_name=operation_name, @@ -222,8 +224,11 @@ async def describe(self) -> ScheduleDescription: class AsyncScheduledTaskClient: """Asynchronous client for managing scheduled tasks.""" - def __init__(self, client: AsyncTaskHubGrpcClient, *, - operation_timeout: float = 60): + def __init__( + self, + client: AsyncTaskHubGrpcClient, + *, + operation_timeout: float = 60): self._client = client self._operation_timeout = operation_timeout @@ -233,7 +238,9 @@ def get_schedule_client(self, schedule_id: str) -> AsyncScheduleClient: self._client, schedule_id, operation_timeout=self._operation_timeout) async def create_schedule( - self, options: ScheduleCreationOptions) -> AsyncScheduleClient: + self, + options: ScheduleCreationOptions, + ) -> AsyncScheduleClient: """Create a new schedule and return a client for managing it.""" schedule_client = self.get_schedule_client(options.schedule_id) await schedule_client.create(options) @@ -247,7 +254,9 @@ async def get_schedule(self, schedule_id: str) -> ScheduleDescription | None: return None async def list_schedules( - self, schedule_query: ScheduleQuery | None = None) -> list[ScheduleDescription]: + self, + schedule_query: ScheduleQuery | None = None, + ) -> list[ScheduleDescription]: """List schedules matching the given filter criteria.""" prefix = schedule_query.schedule_id_prefix if schedule_query and schedule_query.schedule_id_prefix else "" page_size = (schedule_query.page_size if schedule_query and schedule_query.page_size From f7f724b49f0652c1b66d5261ddffb254f9fa264c Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 12:53:56 -0600 Subject: [PATCH 07/14] Handle empty async export state Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- durabletask/extensions/history_export/client.py | 2 +- .../extensions/history_export/test_client.py | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/durabletask/extensions/history_export/client.py b/durabletask/extensions/history_export/client.py index f1e5392f..1ab0ad54 100644 --- a/durabletask/extensions/history_export/client.py +++ b/durabletask/extensions/history_export/client.py @@ -534,7 +534,7 @@ async def get_job(self, job_id: str) -> ExportJobDescription | None: if meta is None: return None state = meta.get_typed_state() - if not isinstance(state, dict): + if not isinstance(state, dict) or not state: return None return ExportJobDescription.from_state_dict(job_id, cast("dict[str, Any]", state)) diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index e853ab48..14388a9c 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -13,6 +13,7 @@ import gzip import json import threading +from unittest.mock import AsyncMock, MagicMock from datetime import datetime, timedelta, timezone import pytest @@ -175,6 +176,16 @@ async def test_async_client_create_list_wait_and_delete( assert await export_client.get_job(desc.job_id) is None +async def test_async_client_get_job_returns_none_for_empty_state(writer): + dt_client = MagicMock() + metadata = MagicMock() + metadata.get_typed_state.return_value = {} + dt_client.get_entity = AsyncMock(return_value=metadata) + export_client = AsyncExportHistoryClient(dt_client, writer) + + assert await export_client.get_job("empty-state") is None + + def test_get_job_returns_none_for_unknown_id(export_client): assert export_client.get_job("does-not-exist") is None From e17a597ffa5dd26fa8e81e8de5933224474e60f6 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 13:12:14 -0600 Subject: [PATCH 08/14] Cover async export job facade Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../extensions/history_export/test_client.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index 14388a9c..73e879fe 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -21,6 +21,7 @@ from durabletask import client, task, worker from durabletask.extensions.history_export import ( AsyncExportHistoryClient, + AsyncExportHistoryJobClient, ExportDestination, ExportFormat, ExportFormatKind, @@ -164,16 +165,21 @@ async def test_async_client_create_list_wait_and_delete( format=ExportFormat(kind=ExportFormatKind.JSON), )) - final = await export_client.wait_for_job( - desc.job_id, timeout=30, poll_interval=0.1) + job_client = export_client.get_job_client(desc.job_id) + assert isinstance(job_client, AsyncExportHistoryJobClient) + assert job_client.job_id == desc.job_id + assert job_client.orchestrator_instance_id == orchestrator_instance_id_for( + desc.job_id) + + final = await job_client.wait(timeout=30, poll_interval=0.1) assert final.status == ExportJobStatus.COMPLETED - assert (await export_client.get_job(desc.job_id)) is not None + assert (await job_client.describe()) is not None assert desc.job_id in { job.job_id async for job in export_client.list_jobs() } - await export_client.delete_job(desc.job_id) - assert await export_client.get_job(desc.job_id) is None + await job_client.delete() + assert await job_client.describe() is None async def test_async_client_get_job_returns_none_for_empty_state(writer): From a6146ae1b21eabcd3684622371ae440a387d94e4 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 13:20:24 -0600 Subject: [PATCH 09/14] Clarify missing client binding errors Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../durable_functions/decorators/durable_app.py | 4 ++++ .../azure-functions-durable/test_decorator_compat.py | 12 ++++++++++++ 2 files changed, 16 insertions(+) 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 27ca73bc..75b29ba6 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -411,6 +411,10 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder: @wraps(function) async def client_bound(*args: Any, **kwargs: Any) -> Any: 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( diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index b74c9ede..8d58da45 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -2,6 +2,7 @@ # Licensed under the MIT License. import azure.durable_functions as df +import pytest from azure.durable_functions.constants import ( ACTIVITY_TRIGGER, DURABLE_CLIENT, @@ -147,6 +148,17 @@ def starter(client): assert clients[0] is clients[1] +async def test_durable_client_input_rejects_missing_binding_parameter(): + app = df.DFApp() + + async def starter(other): + return other + + fb = app.durable_client_input(client_name="client")(starter) + with pytest.raises(TypeError, match="binding parameter 'client' is not declared"): + await fb._function._func(other="{}") + + # --------------------------------------------------------------------------- # All decorators register a function builder # --------------------------------------------------------------------------- From 53fcb26a9386bd6f933210b44b208bd37b88c1e2 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 14:10:44 -0600 Subject: [PATCH 10/14] Align changelog entry style Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- CHANGELOG.md | 6 +++--- azure-functions-durable/CHANGELOG.md | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37758446..d93d86e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,9 +10,9 @@ 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. +`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()` 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. diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index fbe9ca96..c467f71e 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -10,9 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. +`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 From dbebdb45b927964253256790ce904f211ba4458b Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 14:28:39 -0600 Subject: [PATCH 11/14] Harden async client error handling Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../decorators/durable_app.py | 44 ++++++++++++++----- .../test_decorator_compat.py | 19 ++++++-- .../extensions/history_export/test_client.py | 42 ++++++++++++++++++ .../test_client_filter_and_capability.py | 35 ++++++++++++++- 4 files changed, 124 insertions(+), 16 deletions(-) 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 75b29ba6..7652cdba 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -408,8 +408,10 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder: if isinstance(user_fn, FunctionBuilder) else user_fn) signature = inspect.signature(function) - @wraps(function) - async def client_bound(*args: Any, **kwargs: Any) -> Any: + 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( @@ -422,15 +424,35 @@ async def client_bound(*args: Any, **kwargs: Any) -> Any: client = (SyncDurableFunctionsClient.get_cached(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 isinstance(client, DurableFunctionsClient): - client.schedule_close() - - client_bound.__annotations__[client_name] = str - setattr(client_bound, "client_function", function) + return bound, client + + def set_client_metadata(client_bound: Callable[..., Any]) -> None: + annotations = dict(function.__annotations__) + annotations.setdefault(client_name, str) + client_bound.__annotations__ = annotations + setattr(client_bound, "client_function", function) + + if inspect.iscoroutinefunction(function): + @wraps(function) + async def 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() + else: + @wraps(function) + def 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() + + set_client_metadata(client_bound) if isinstance(user_fn, FunctionBuilder): user_fn._function._func = client_bound # pyright: ignore[reportPrivateUsage] return wrap(user_fn) diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index 8d58da45..bb6ced27 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -2,6 +2,7 @@ # Licensed under the MIT License. import azure.durable_functions as df +import inspect import pytest from azure.durable_functions.constants import ( ACTIVITY_TRIGGER, @@ -134,7 +135,7 @@ async def starter(client): assert await fb._function._func(client="{}") == "DurableFunctionsClient" -async def test_durable_client_input_sync_injects_sync_client(): +def test_durable_client_input_sync_injects_sync_client(): app = df.DFApp() clients = [] @@ -143,8 +144,9 @@ def starter(client): return type(client).__name__ fb = app.durable_client_input_sync(client_name="client")(starter) - assert await fb._function._func(client="{}") == "SyncDurableFunctionsClient" - assert await fb._function._func(client="{}") == "SyncDurableFunctionsClient" + assert not inspect.iscoroutinefunction(fb._function._func) + assert fb._function._func(client="{}") == "SyncDurableFunctionsClient" + assert fb._function._func(client="{}") == "SyncDurableFunctionsClient" assert clients[0] is clients[1] @@ -159,6 +161,17 @@ async def starter(other): await fb._function._func(other="{}") +async def test_durable_client_input_preserves_client_annotation(): + app = df.DFApp() + + async def starter(client: df.DurableFunctionsClient): + return type(client).__name__ + + fb = app.durable_client_input(client_name="client")(starter) + assert starter.__annotations__["client"] is df.DurableFunctionsClient + assert await fb._function._func(client="{}") == "DurableFunctionsClient" + + # --------------------------------------------------------------------------- # All decorators register a function builder # --------------------------------------------------------------------------- diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index 73e879fe..aff3c1a1 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -17,6 +17,7 @@ from datetime import datetime, timedelta, timezone import pytest +import grpc from durabletask import client, task, worker from durabletask.extensions.history_export import ( @@ -192,6 +193,47 @@ async def test_async_client_get_job_returns_none_for_empty_state(writer): assert await export_client.get_job("empty-state") is None +async def test_async_wait_for_job_raises_lookup_when_job_never_exists(writer): + export_client = AsyncExportHistoryClient(MagicMock(), writer) + export_client.get_job = AsyncMock(return_value=None) + + with pytest.raises(ExportJobNotFoundError): + await export_client.wait_for_job( + "never-created", timeout=0.01, poll_interval=0.001) + + +async def test_async_wait_for_job_times_out_when_status_stays_active(writer): + export_client = AsyncExportHistoryClient(MagicMock(), writer) + export_client.get_job = AsyncMock( + return_value=MagicMock(status=ExportJobStatus.ACTIVE)) + + with pytest.raises(TimeoutError, match="did not reach a terminal status"): + await export_client.wait_for_job( + "stuck-job", timeout=0.01, poll_interval=0.001) + + +class _NotFoundRpcError(grpc.RpcError): + def code(self): + return grpc.StatusCode.NOT_FOUND + + +async def test_async_delete_job_ignores_not_found_errors(writer): + dt_client = MagicMock() + dt_client.signal_entity = AsyncMock() + dt_client.terminate_orchestration = AsyncMock(side_effect=_NotFoundRpcError()) + dt_client.wait_for_orchestration_completion = AsyncMock( + side_effect=_NotFoundRpcError()) + dt_client.purge_orchestration = AsyncMock(side_effect=_NotFoundRpcError()) + export_client = AsyncExportHistoryClient(dt_client, writer) + + await export_client.delete_job("missing-job") + + dt_client.signal_entity.assert_awaited_once() + dt_client.terminate_orchestration.assert_awaited_once() + dt_client.wait_for_orchestration_completion.assert_awaited_once() + dt_client.purge_orchestration.assert_awaited_once() + + def test_get_job_returns_none_for_unknown_id(export_client): assert export_client.get_job("does-not-exist") is None diff --git a/tests/durabletask/scheduled/test_client_filter_and_capability.py b/tests/durabletask/scheduled/test_client_filter_and_capability.py index 5533ce15..4de3ccbd 100644 --- a/tests/durabletask/scheduled/test_client_filter_and_capability.py +++ b/tests/durabletask/scheduled/test_client_filter_and_capability.py @@ -4,9 +4,13 @@ """Unit tests for ScheduleClient filtering and scheduled-tasks worker capability.""" from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock +import pytest import durabletask.internal.orchestrator_service_pb2 as pb -from durabletask.scheduled.client import ScheduledTaskClient +from durabletask.client import OrchestrationStatus +from durabletask.scheduled.client import AsyncScheduleClient, ScheduledTaskClient from durabletask.scheduled.models import (ScheduleConfiguration, ScheduleCreationOptions, ScheduleQuery, ScheduleState) @@ -76,8 +80,35 @@ def test_capability_absent_by_default(self): assert pb.WORKER_CAPABILITY_SCHEDULED_TASKS not in worker._capabilities # pyright: ignore[reportPrivateUsage] def test_add_capability_rejected_while_running(self): - import pytest worker = TaskHubGrpcWorker() worker._is_running = True # pyright: ignore[reportPrivateUsage] with pytest.raises(RuntimeError): worker.add_capability(pb.WORKER_CAPABILITY_SCHEDULED_TASKS) + + +async def test_async_schedule_operation_raises_failure_details(): + client = MagicMock() + client.schedule_new_orchestration = AsyncMock(return_value="operation-id") + client.wait_for_orchestration_completion = AsyncMock(return_value=SimpleNamespace( + runtime_status=OrchestrationStatus.FAILED, + failure_details=SimpleNamespace(message="operation failed"), + )) + schedule = AsyncScheduleClient(client, "schedule-id") + + with pytest.raises(RuntimeError, match="operation failed"): + await schedule.create(ScheduleCreationOptions( + schedule_id="schedule-id", + orchestration_name="orchestration", + interval=timedelta(minutes=1), + )) + + +async def test_async_schedule_operation_propagates_timeout(): + client = MagicMock() + client.schedule_new_orchestration = AsyncMock(return_value="operation-id") + client.wait_for_orchestration_completion = AsyncMock( + side_effect=TimeoutError("operation timed out")) + schedule = AsyncScheduleClient(client, "schedule-id") + + with pytest.raises(TimeoutError, match="operation timed out"): + await schedule.pause() From 524ddb2ce98b1c21150244dba97ba01ca4fbf987 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 14:31:27 -0600 Subject: [PATCH 12/14] Avoid client wrapper redeclaration Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../azure/durable_functions/decorators/durable_app.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 7652cdba..025a90fa 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -434,7 +434,7 @@ def set_client_metadata(client_bound: Callable[..., Any]) -> None: if inspect.iscoroutinefunction(function): @wraps(function) - async def client_bound(*args: Any, **kwargs: Any) -> Any: + async def async_client_bound(*args: Any, **kwargs: Any) -> Any: bound, client = bind_client(args, kwargs) try: result = function(*bound.args, **bound.kwargs) @@ -442,15 +442,17 @@ async def client_bound(*args: Any, **kwargs: Any) -> Any: finally: if isinstance(client, DurableFunctionsClient): client.schedule_close() + client_bound = async_client_bound else: @wraps(function) - def client_bound(*args: Any, **kwargs: Any) -> Any: + 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): From 8c5a62655f040e6aeaa437d8df1538d211abe45b Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 14:36:29 -0600 Subject: [PATCH 13/14] Preserve supported client annotations Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- .../azure/durable_functions/decorators/durable_app.py | 9 ++++++++- .../azure-functions-durable/test_decorator_compat.py | 11 +++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) 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 025a90fa..6adbd918 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -428,7 +428,14 @@ def bind_client( def set_client_metadata(client_bound: Callable[..., Any]) -> None: annotations = dict(function.__annotations__) - annotations.setdefault(client_name, str) + 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) diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index bb6ced27..02acef04 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -172,6 +172,17 @@ async def starter(client: df.DurableFunctionsClient): assert await fb._function._func(client="{}") == "DurableFunctionsClient" +async def test_durable_client_input_replaces_unsupported_client_annotation(): + app = df.DFApp() + + async def starter(client: int): + return type(client).__name__ + + fb = app.durable_client_input(client_name="client")(starter) + assert fb._function._func.__annotations__["client"] is str + assert starter.__annotations__["client"] is int + + # --------------------------------------------------------------------------- # All decorators register a function builder # --------------------------------------------------------------------------- From 6dc3abed4421c61c08a84de7b526eb20c67042fb Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 27 Jul 2026 14:40:54 -0600 Subject: [PATCH 14/14] Infer Functions durable client type Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 76f2163a-0773-4889-b627-1bd4e5c7bea0 --- azure-functions-durable/CHANGELOG.md | 8 ++--- .../decorators/durable_app.py | 34 +++++++------------ durabletask/scheduled/client.py | 7 ++-- .../test_decorator_compat.py | 4 +-- 4 files changed, 23 insertions(+), 30 deletions(-) diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index c467f71e..5bea7251 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,10 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. +- 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 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 6adbd918..17a5664b 100644 --- a/azure-functions-durable/azure/durable_functions/decorators/durable_app.py +++ b/azure-functions-durable/azure/durable_functions/decorators/durable_app.py @@ -152,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", sync=True)( + self.durable_client_input(client_name="client")( self.activity_trigger( input_name="input", activity=LIST_TERMINAL_INSTANCES_ACTIVITY)( list_terminal_instances_client_bound)) - self.durable_client_input(client_name="client", sync=True)( + self.durable_client_input(client_name="client")( self.activity_trigger( input_name="input", activity=EXPORT_INSTANCE_HISTORY_ACTIVITY)( @@ -359,11 +359,13 @@ def durable_client_input(self, client_name: str, task_hub: Optional[str] = None, connection_name: Optional[str] = None, - *, - sync: bool = False, ) -> 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 @@ -378,17 +380,14 @@ 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 def wrap(fb: FunctionBuilder) -> FunctionBuilder: def decorator() -> FunctionBuilder: # The converter returns the host configuration string. The - # function wrapper below constructs and closes the requested - # synchronous or asynchronous rich client per invocation. + # 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, @@ -407,6 +406,7 @@ def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder: 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, ...], @@ -421,8 +421,8 @@ def bind_client( if not isinstance(raw_client, str): raise TypeError( f"durable client binding '{client_name}' did not provide its configuration") - client = (SyncDurableFunctionsClient.get_cached(raw_client) if sync - else DurableFunctionsClient(raw_client)) + client = (DurableFunctionsClient(raw_client) if is_async_function + else SyncDurableFunctionsClient.get_cached(raw_client)) bound.arguments[client_name] = client return bound, client @@ -439,7 +439,7 @@ def set_client_metadata(client_bound: Callable[..., Any]) -> None: client_bound.__annotations__ = annotations setattr(client_bound, "client_function", function) - if inspect.iscoroutinefunction(function): + if is_async_function: @wraps(function) async def async_client_bound(*args: Any, **kwargs: Any) -> Any: bound, client = bind_client(args, kwargs) @@ -469,16 +469,6 @@ def sync_client_bound(*args: Any, **kwargs: Any) -> Any: 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. diff --git a/durabletask/scheduled/client.py b/durabletask/scheduled/client.py index 819dcb0a..c3e7a3a8 100644 --- a/durabletask/scheduled/client.py +++ b/durabletask/scheduled/client.py @@ -269,7 +269,10 @@ async def list_schedules( results: list[ScheduleDescription] = [] for metadata in await self._client.get_all_entities(query): state = metadata.get_typed_state(ScheduleState) - if state is not None and state.schedule_configuration is not None and \ - ScheduledTaskClient.matches_filter(state, schedule_query): + if ( + state is not None + and state.schedule_configuration is not None + and ScheduledTaskClient.matches_filter(state, schedule_query) + ): results.append(state.to_description()) return results diff --git a/tests/azure-functions-durable/test_decorator_compat.py b/tests/azure-functions-durable/test_decorator_compat.py index 02acef04..898a277c 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -135,7 +135,7 @@ async def starter(client): assert await fb._function._func(client="{}") == "DurableFunctionsClient" -def test_durable_client_input_sync_injects_sync_client(): +def test_durable_client_input_injects_sync_client_for_sync_function(): app = df.DFApp() clients = [] @@ -143,7 +143,7 @@ def starter(client): clients.append(client) return type(client).__name__ - fb = app.durable_client_input_sync(client_name="client")(starter) + fb = app.durable_client_input(client_name="client")(starter) assert not inspect.iscoroutinefunction(fb._function._func) assert fb._function._func(client="{}") == "SyncDurableFunctionsClient" assert fb._function._func(client="{}") == "SyncDurableFunctionsClient"