diff --git a/CHANGELOG.md b/CHANGELOG.md index 98444e19..d93d86e4 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..5bea7251 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`. `DFApp.durable_client_input()` now injects +the synchronous client into synchronous functions and the asynchronous client +into coroutine functions. Both clients support scheduled-task and history-export +APIs without an async-to-sync bridge. + ## 2.0.0b1 First preview (beta) release of `azure-functions-durable` 2.x — a ground-up 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..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 @@ -14,10 +16,14 @@ AsyncTaskHubGrpcClient, OrchestrationQuery, OrchestrationStatus, + TaskHubGrpcClient, ) from durabletask.entities import EntityInstanceId from durabletask.grpc_options import GrpcChannelOptions -from .internal.azurefunctions_grpc_interceptor import AzureFunctionsAsyncDefaultClientInterceptorImpl +from .internal.azurefunctions_grpc_interceptor import ( + AzureFunctionsAsyncDefaultClientInterceptorImpl, + AzureFunctionsDefaultClientInterceptorImpl, +) from .internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER from .http import HttpManagementPayload from .internal.compat.durable_orchestration_status import DurableOrchestrationStatus @@ -26,6 +32,10 @@ from .internal.compat.purge_history_result import PurgeHistoryResult +_sync_client_cache: dict[str, "SyncDurableFunctionsClient"] = {} +_sync_client_cache_lock = threading.Lock() + + # Client class used for Durable Functions class DurableFunctionsClient(AsyncTaskHubGrpcClient): """A gRPC client passed to Durable Functions durable client bindings. @@ -490,3 +500,121 @@ def _create_http_response(status_code: int, body: Union[str, Any]) -> func.HttpR body=body_as_json, mimetype="application/json", headers={"Content-Type": "application/json"}) + + +class SyncDurableFunctionsClient(TaskHubGrpcClient): + """Synchronous durable client supplied by a Functions durable-client binding.""" + + taskHubName: str + connectionName: str + creationUrls: dict[str, str] + managementUrls: dict[str, str] + baseUrl: str + requiredQueryStringParameters: str + rpcBaseUrl: str + httpBaseUrl: str + maxGrpcMessageSizeInBytes: int + grpcHttpClientTimeout: timedelta | str + + def __init__(self, client_as_string: str): + self._parse_client_configuration(client_as_string) + interceptors = [AzureFunctionsDefaultClientInterceptorImpl( + self.taskHubName, self.requiredQueryStringParameters)] + channel_options: GrpcChannelOptions | None = None + if self.maxGrpcMessageSizeInBytes > 0: + channel_options = GrpcChannelOptions( + max_receive_message_length=self.maxGrpcMessageSizeInBytes, + max_send_message_length=self.maxGrpcMessageSizeInBytes) + super().__init__( + host_address=self.rpcBaseUrl, + secure_channel=False, + metadata=None, + interceptors=interceptors, + channel_options=channel_options, + data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) + + @classmethod + def get_cached(cls, client_as_string: str) -> "SyncDurableFunctionsClient": + """Get the process-wide client for a durable-client binding configuration. + + Synchronous Functions bindings can be invoked frequently by history + export fan-out activities. Reusing the gRPC channel avoids creating and + tearing down a channel for every activity invocation. + """ + with _sync_client_cache_lock: + cached = _sync_client_cache.get(client_as_string) + if cached is None: + cached = cls(client_as_string) + _sync_client_cache[client_as_string] = cached + return cached + + def _parse_client_configuration(self, client_as_string: str) -> None: + client = json.loads(client_as_string) + self.taskHubName = client.get("taskHubName") or "" + self.connectionName = client.get("connectionName") or "" + self.creationUrls = client.get("creationUrls") or {} + self.managementUrls = client.get("managementUrls") or {} + self.baseUrl = client.get("baseUrl") or "" + self.requiredQueryStringParameters = client.get( + "requiredQueryStringParameters") or "" + self.rpcBaseUrl = client.get("rpcBaseUrl") or "" + self.httpBaseUrl = client.get("httpBaseUrl") or "" + self.maxGrpcMessageSizeInBytes = client.get( + "maxGrpcMessageSizeInBytes") or 0 + self.grpcHttpClientTimeout = client.get( + "grpcHttpClientTimeout") or timedelta(seconds=30) + + def create_check_status_response( + self, request: func.HttpRequest, instance_id: str) -> func.HttpResponse: + payload = self._get_client_response_links(request, instance_id) + return func.HttpResponse( + body=str(payload), + status_code=202, + headers={ + "content-type": "application/json", + "Location": payload["statusQueryGetUri"], + }, + ) + + def create_http_management_payload( + self, + request: func.HttpRequest | str | None = None, + instance_id: str | None = None) -> HttpManagementPayload: + if instance_id is None and isinstance(request, str): + instance_id = request + request = None + if instance_id is None: + raise TypeError("instance_id is required") + resolved_request = request if isinstance(request, func.HttpRequest) else None + return self._get_client_response_links(resolved_request, instance_id) + + def _get_client_response_links( + self, request: func.HttpRequest | None, + instance_id: str) -> HttpManagementPayload: + return HttpManagementPayload( + instance_id, + self._get_instance_status_url(request, instance_id), + self.requiredQueryStringParameters) + + def _get_instance_status_url( + self, request: func.HttpRequest | None, instance_id: str) -> str: + encoded_instance_id = quote(instance_id) + if request is not None: + request_url = urlparse(request.url) + return ( + f"{request_url.scheme}://{request_url.netloc}" + f"/runtime/webhooks/durabletask/instances/{encoded_instance_id}") + base_url = self.baseUrl.rstrip("/") if self.baseUrl else "" + return f"{base_url}/instances/{encoded_instance_id}" + + +def _close_cached_sync_clients() -> None: + """Release process-wide synchronous durable-client channels on shutdown.""" + with _sync_client_cache_lock: + clients = list(_sync_client_cache.values()) + _sync_client_cache.clear() + for client in clients: + client.close() + + +atexit.register(_close_cached_sync_clients) 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..17a5664b 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 @@ -356,10 +358,14 @@ def decorator() -> FunctionBuilder: def durable_client_input(self, client_name: str, task_hub: Optional[str] = None, - connection_name: Optional[str] = None + connection_name: Optional[str] = None, ) -> Callable[[Callable[..., Any]], FunctionBuilder]: """Register a Durable-client Function. + Coroutine functions receive an asynchronous + :class:`DurableFunctionsClient`; synchronous functions receive a cached + :class:`SyncDurableFunctionsClient`. + Parameters ---------- client_name: str @@ -379,11 +385,9 @@ def durable_client_input(self, @self._build_function def wrap(fb: FunctionBuilder) -> FunctionBuilder: def decorator() -> FunctionBuilder: - # The durableClient binding converter - # (``DurableClientConverter``) constructs the rich - # ``DurableFunctionsClient`` from the host-supplied configuration - # string during decode, so no per-function client middleware is - # needed here. + # The converter returns the host configuration string. The + # function wrapper below constructs the rich client matching + # the decorated function's synchronous or asynchronous type. fb.add_binding( binding=DurableClient(name=client_name, task_hub=task_hub, @@ -393,16 +397,75 @@ def decorator() -> FunctionBuilder: return decorator() def attach_client_function(user_fn: Callable[..., Any]) -> FunctionBuilder: - # Expose the raw client function for unit testing, mirroring the - # ``.orchestrator_function`` and ``.entity_function`` attributes. - # Unlike v1 there is no client middleware wrapper (the durableClient - # binding converter builds the rich client during decode), so the - # user function is registered directly and ``.client_function`` - # points back at it. Internal callers pass an already-built - # ``FunctionBuilder`` (e.g. history export), which is left untouched. - if not isinstance(user_fn, FunctionBuilder): - user_fn.client_function = user_fn # pyright: ignore[reportFunctionMemberAccess] - return wrap(user_fn) + # Expose the original client function for unit testing, mirroring + # ``.orchestrator_function`` and ``.entity_function``. The registered + # wrapper resolves the client from the binding configuration and + # closes it after the invocation. + from ..client import DurableFunctionsClient, SyncDurableFunctionsClient + + function = (user_fn._function._func # pyright: ignore[reportPrivateUsage] + if isinstance(user_fn, FunctionBuilder) else user_fn) + signature = inspect.signature(function) + is_async_function = inspect.iscoroutinefunction(function) + + def bind_client( + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> tuple[inspect.BoundArguments, DurableFunctionsClient | SyncDurableFunctionsClient]: + bound = signature.bind(*args, **kwargs) + if client_name not in bound.arguments: + raise TypeError( + f"durable client binding parameter '{client_name}' is not " + f"declared by function '{function.__name__}'") + raw_client = bound.arguments[client_name] + if not isinstance(raw_client, str): + raise TypeError( + f"durable client binding '{client_name}' did not provide its configuration") + client = (DurableFunctionsClient(raw_client) if is_async_function + else SyncDurableFunctionsClient.get_cached(raw_client)) + bound.arguments[client_name] = client + return bound, client + + def set_client_metadata(client_bound: Callable[..., Any]) -> None: + annotations = dict(function.__annotations__) + supported_client_types = ( + str, + bytes, + DurableFunctionsClient, + SyncDurableFunctionsClient, + ) + if annotations.get(client_name) not in supported_client_types: + annotations[client_name] = str + client_bound.__annotations__ = annotations + setattr(client_bound, "client_function", function) + + if is_async_function: + @wraps(function) + async def async_client_bound(*args: Any, **kwargs: Any) -> Any: + bound, client = bind_client(args, kwargs) + try: + result = function(*bound.args, **bound.kwargs) + return await result + finally: + if isinstance(client, DurableFunctionsClient): + client.schedule_close() + client_bound = async_client_bound + else: + @wraps(function) + def sync_client_bound(*args: Any, **kwargs: Any) -> Any: + bound, client = bind_client(args, kwargs) + try: + return function(*bound.args, **bound.kwargs) + finally: + if isinstance(client, DurableFunctionsClient): + client.schedule_close() + client_bound = sync_client_bound + + set_client_metadata(client_bound) + if isinstance(user_fn, FunctionBuilder): + user_fn._function._func = client_bound # pyright: ignore[reportPrivateUsage] + return wrap(user_fn) + return wrap(client_bound) return attach_client_function diff --git a/azure-functions-durable/azure/durable_functions/internal/converters.py b/azure-functions-durable/azure/durable_functions/internal/converters.py index 09de202f..c0892f4c 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,16 +214,19 @@ def encode(cls, obj: Any, *, @classmethod def decode(cls, data: meta.Datum, *, trigger_metadata: _TriggerMetadata) -> Any: - from ..client import DurableFunctionsClient - return DurableFunctionsClient(data.value) + # The decorator turns this host configuration into the requested sync or + # async rich client. Keeping conversion here as a string lets one binding + # type support both client shapes without a hidden sync bridge. + return data.value def register_durable_converters() -> None: """Register this package's durable binding converters with azure-functions. Overrides the SDK's built-in converters for the four durable binding types - so the host uses the durabletask-based encodings and the durable-client - binding is decoded to a :class:`DurableFunctionsClient`. + so the host uses the durabletask-based encodings. The durable-client binding + is decoded to its host configuration string, which the decorator converts to + the requested synchronous or asynchronous rich client per invocation. """ func.register_converter( ORCHESTRATION_TRIGGER, OrchestrationTriggerConverter, overwrite=True) 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..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,8 +20,6 @@ from __future__ import annotations -import atexit -import threading from collections.abc import Mapping from datetime import datetime, timezone from typing import Any, Optional, cast @@ -49,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. @@ -157,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: @@ -175,72 +164,13 @@ 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: 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..1ab0ad54 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,196 @@ 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) or not state: + 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..bfdb1549 100644 --- a/durabletask/scheduled/__init__.py +++ b/durabletask/scheduled/__init__.py @@ -6,10 +6,13 @@ 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 ScheduleClient, ScheduledTaskClient +from durabletask.scheduled.client import (AsyncScheduleClient, + AsyncScheduledTaskClient, + ScheduleClient, ScheduledTaskClient) from durabletask.scheduled.exceptions import (ScheduleClientValidationError, ScheduleError, ScheduleInvalidTransitionError, @@ -21,7 +24,9 @@ __all__ = [ "ScheduledTaskClient", + "AsyncScheduledTaskClient", "ScheduleClient", + "AsyncScheduleClient", "ScheduleCreationOptions", "ScheduleUpdateOptions", "ScheduleDescription", diff --git a/durabletask/scheduled/client.py b/durabletask/scheduled/client.py index dc6767ac..c3e7a3a8 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 @@ -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: @@ -152,3 +152,127 @@ 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..898a277c 100644 --- a/tests/azure-functions-durable/test_decorator_compat.py +++ b/tests/azure-functions-durable/test_decorator_compat.py @@ -2,6 +2,8 @@ # Licensed under the MIT License. import azure.durable_functions as df +import inspect +import pytest from azure.durable_functions.constants import ( ACTIVITY_TRIGGER, DURABLE_CLIENT, @@ -123,17 +125,62 @@ 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 await fb._function._func(client="{}") == "DurableFunctionsClient" + + +def test_durable_client_input_injects_sync_client_for_sync_function(): + app = df.DFApp() + clients = [] + + def starter(client): + clients.append(client) + return type(client).__name__ + + 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" + 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="{}") + + +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" + + +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 is starter + assert fb._function._func.__annotations__["client"] is str + assert starter.__annotations__["client"] is int # --------------------------------------------------------------------------- diff --git a/tests/azure-functions-durable/test_history_export_compat.py b/tests/azure-functions-durable/test_history_export_compat.py index 74521cfa..e1a67c6d 100644 --- a/tests/azure-functions-durable/test_history_export_compat.py +++ b/tests/azure-functions-durable/test_history_export_compat.py @@ -189,53 +189,21 @@ 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(fake_client) - context = hec._context_for(object()) - - # 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): - 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 diff --git a/tests/durabletask/extensions/history_export/test_client.py b/tests/durabletask/extensions/history_export/test_client.py index 3275b50f..aff3c1a1 100644 --- a/tests/durabletask/extensions/history_export/test_client.py +++ b/tests/durabletask/extensions/history_export/test_client.py @@ -13,12 +13,16 @@ import gzip import json import threading +from unittest.mock import AsyncMock, MagicMock from datetime import datetime, timedelta, timezone import pytest +import grpc from durabletask import client, task, worker from durabletask.extensions.history_export import ( + AsyncExportHistoryClient, + AsyncExportHistoryJobClient, ExportDestination, ExportFormat, ExportFormatKind, @@ -148,6 +152,88 @@ 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), + )) + + 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 job_client.describe()) is not None + assert desc.job_id in { + job.job_id async for job in export_client.list_jobs() + } + + await job_client.delete() + assert await job_client.describe() 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 + + +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 66cfc318..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) @@ -30,7 +34,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 +42,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 +56,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: @@ -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() 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()