diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 5acd7412..5be6c6c6 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -43,7 +43,8 @@ Examples: ## Language and Style -- Python 3.10+ is required. +- Python 3.11+ is required for the core and Azure Managed packages. +- Python 3.13+ is required for the Azure Functions Durable package. - Use type hints for all public API signatures. - Follow PEP 8 conventions. - Use `autopep8` for Python formatting. diff --git a/.github/workflows/durabletask-azuremanaged.yml b/.github/workflows/durabletask-azuremanaged.yml index ff06674d..6719b966 100644 --- a/.github/workflows/durabletask-azuremanaged.yml +++ b/.github/workflows/durabletask-azuremanaged.yml @@ -33,7 +33,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"] + python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"] needs: lint runs-on: ubuntu-latest steps: diff --git a/.github/workflows/durabletask.yml b/.github/workflows/durabletask.yml index 5da31933..e52b9fc6 100644 --- a/.github/workflows/durabletask.yml +++ b/.github/workflows/durabletask.yml @@ -33,7 +33,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"] + python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"] needs: lint-and-unit-tests runs-on: ubuntu-latest steps: diff --git a/.github/workflows/typecheck.yml b/.github/workflows/typecheck.yml index 7b10ed31..a577b406 100644 --- a/.github/workflows/typecheck.yml +++ b/.github/workflows/typecheck.yml @@ -22,17 +22,17 @@ jobs: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - - name: Set up Python 3.10 (lowest supported) + - name: Set up Python 3.11 (lowest supported) uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 with: - python-version: "3.10" + python-version: "3.11" - name: Install packages and dependencies run: | python -m pip install --upgrade pip pip install nox - - name: Run pyright (strict, Python 3.10) + - name: Run pyright (strict, Python 3.11) run: nox -s typecheck_core pyright-azurefunctions: diff --git a/CHANGELOG.md b/CHANGELOG.md index 56778154..864adff1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,33 @@ ADDED - Added `OrchestrationQuery.task_hub_names` to filter orchestration instances by one or more task hubs. +CHANGED + +- Dropped Python 3.10 support. Python 3.11 or later is now required. +- Removed the obsolete PyPI `asyncio` backport dependency; all supported +Python versions use the standard-library module. +- Context managers now preserve subclass types in type checkers, including +Azure Managed clients and workers. +- Entity ID copy and pickle hooks now preserve subclass types in type checkers. +- Multiple task completion callback failures now raise the built-in +`ExceptionGroup`, preserving each exception and its traceback in memory and +the first callback error as the explicit cause and serialized inner failure. +A single callback failure is still raised directly. + +> [!WARNING] +> The aggregate exception type and message have changed. Orchestrations that +> branch on the old aggregate type or message can take a different path during +> replay; drain affected instances on the previous SDK or use a separate task +> hub for new instances. Branches on the aggregate's explicit first-error cause +> retain their previous behavior. + FIXED +- Grouped failures now retain nested exception messages, tracebacks, and notes +in failure details, including `ExceptionGroup` failures from task callbacks. +Traceback rendering is limited to 15 children per group and 10 group levels, +with truncation markers for omitted diagnostics. Failure details do not +serialize a complete structured tree of group children. - Optimized set and frozenset serialization, particularly for deeply nested frozensets and large singleton elements. JSON object names within sets are now ordered consistently after conversion to strings, including numeric names. diff --git a/README.md b/README.md index 2bd8e7fb..d37c5d97 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,11 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/ - [`azure-functions-durable`](./azure-functions-durable/) is a preview Azure Durable Functions provider built on `durabletask`. +## Python Requirements + +The core SDK and Azure Managed provider require Python 3.11 or later. +The Azure Functions Durable provider requires Python 3.13 or later. + ## References - [Supported Patterns](./docs/supported-patterns.md) diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 06b5ef64..70360b44 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -18,9 +18,21 @@ The opt-in uses the shared codec's `ensure_ascii` option available in CHANGED - Updated the minimum `azure-functions` dependency to v2.4.0b1. +- With the corresponding core SDK update, client and worker context managers +preserve Functions subclass types in type checkers. Python 3.13 remains the +minimum supported version for this package. +- With the corresponding core SDK update, multiple task completion callback +failures raise the built-in `ExceptionGroup`; a single failure is still raised +directly. The first callback error remains the explicit cause and serialized +inner failure. Branches on the previous aggregate exception type or message +can change during replay; see the [core upgrade warning](../CHANGELOG.md#unreleased). FIXED +- With the corresponding core SDK update, grouped failures retain nested +exception messages, tracebacks, and notes in failure details. Tracebacks are +limited to 15 children per group and 10 group levels with truncation markers; +group children are not serialized as a complete structured tree. - With the corresponding core `durabletask` SDK fix, oversized messages fail large-payload validation before any uploads or payload replacements. Size errors now take precedence over hypothetical earlier storage failures. diff --git a/azure-functions-durable/azure/durable_functions/http/builtin.py b/azure-functions-durable/azure/durable_functions/http/builtin.py index b921e772..0fe4ff42 100644 --- a/azure-functions-durable/azure/durable_functions/http/builtin.py +++ b/azure-functions-durable/azure/durable_functions/http/builtin.py @@ -29,7 +29,7 @@ import urllib.error import urllib.request -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from email.utils import parsedate_to_datetime from typing import Any, Generator, Optional from urllib.parse import urljoin, urlparse @@ -258,9 +258,9 @@ def _retry_after_seconds(headers: dict[str, str], now: datetime) -> int: # header); treat it as UTC. Normalize ``now`` the same way so the # subtraction is well-defined regardless of the caller's tz-awareness. if retry_at.tzinfo is None: - retry_at = retry_at.replace(tzinfo=timezone.utc) + retry_at = retry_at.replace(tzinfo=UTC) if now.tzinfo is None: - now = now.replace(tzinfo=timezone.utc) + now = now.replace(tzinfo=UTC) return max(int((retry_at - now).total_seconds()), 0) diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py index 17f6b2b5..07ca815c 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/durable_orchestration_status.py @@ -2,7 +2,7 @@ # Licensed under the MIT License. import json -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, Optional, cast from durabletask.client import OrchestrationState, OrchestrationStatus @@ -211,5 +211,5 @@ def _format_datetime(value: datetime) -> str: consumers that parse with that exact format working. """ if value.tzinfo is not None: - value = value.astimezone(timezone.utc).replace(tzinfo=None) + value = value.astimezone(UTC).replace(tzinfo=None) return value.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py index fcf6ea1d..ce0e8440 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/history_projection.py @@ -3,7 +3,7 @@ import json from dataclasses import fields, is_dataclass -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, cast from durabletask import history, task @@ -185,5 +185,5 @@ def _raw_payload(serialized: str) -> Any: def _format_datetime(value: datetime) -> str: if value.tzinfo is not None: - value = value.astimezone(timezone.utc).replace(tzinfo=None) + value = value.astimezone(UTC).replace(tzinfo=None) return value.strftime("%Y-%m-%dT%H:%M:%S.%f") + "Z" diff --git a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py index 4dccb98a..f93176dd 100644 --- a/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py +++ b/azure-functions-durable/azure/durable_functions/internal/compat/orchestration_context.py @@ -3,7 +3,7 @@ import inspect import json -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, Callable, Generator, Optional, cast from uuid import UUID @@ -74,7 +74,7 @@ def current_utc_datetime(self) -> datetime: """ value = self._ctx.current_utc_datetime if value.tzinfo is None: - return value.replace(tzinfo=timezone.utc) + return value.replace(tzinfo=UTC) return value @property 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 2bf7d0bb..1bb72a34 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 @@ -21,7 +21,7 @@ from __future__ import annotations from collections.abc import Mapping -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, Optional, cast from durabletask.client import ( @@ -217,7 +217,7 @@ def export_instance_history_client_bound( def _created_at_from(payload: Mapping[str, Any]) -> datetime: raw = payload.get("created_at") parsed = dt_from_iso(raw) if raw else None - return parsed or datetime.now(timezone.utc) + return parsed or datetime.now(UTC) class FunctionsExportJobEntity(ExportJobEntity): diff --git a/azure-functions-durable/azure/durable_functions/testing/entity.py b/azure-functions-durable/azure/durable_functions/testing/entity.py index b3a2e9e6..dc1acc67 100644 --- a/azure-functions-durable/azure/durable_functions/testing/entity.py +++ b/azure-functions-durable/azure/durable_functions/testing/entity.py @@ -7,7 +7,7 @@ import inspect from dataclasses import dataclass, field -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, Callable, TypeAlias, TypeVar, overload from durabletask.entities import DurableEntity, EntityContext, EntityInstanceId @@ -228,7 +228,7 @@ def _decode_action( if action.HasField("sendSignal"): signal = action.sendSignal scheduled_time = ( - signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) + signal.scheduledTime.ToDatetime(tzinfo=UTC) if signal.HasField("scheduledTime") else None ) diff --git a/docs/development.md b/docs/development.md index bed55819..3c852ab7 100644 --- a/docs/development.md +++ b/docs/development.md @@ -12,10 +12,10 @@ nox -s ci ``` The `ci` session runs linting, strict type checks, core SDK and Azure Managed -tests on Python 3.10, Azure Functions unit tests on Python 3.13, and Azure +tests on Python 3.11, Azure Functions unit tests on Python 3.13, and Azure Functions end-to-end tests on Python 3.13. It intentionally runs one representative version rather than the complete CI matrix, keeping routine -local validation fast. To use another supported version (3.10--3.15) for the +local validation fast. To use another supported version (3.11--3.15) for the core SDK and Azure Managed tests, pass it after `--`: ```sh @@ -53,8 +53,8 @@ For a focused change, run just the relevant session: nox -R -s lint -- durabletask/client.py tests/durabletask/test_client.py nox -R -s typecheck_core -- durabletask/client.py durabletask/extensions/history_export nox -R -s typecheck_functions -- azure-functions-durable/azure -nox -R -s core_tests-3.10 -- tests/durabletask/test_client.py::test_get_grpc_channel_insecure -nox -R -s azuremanaged_tests-3.10 -- tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py +nox -R -s core_tests-3.11 -- tests/durabletask/test_client.py::test_get_grpc_channel_insecure +nox -R -s azuremanaged_tests-3.11 -- tests/durabletask-azuremanaged/test_dts_orchestration_e2e.py nox -R -s functions_unit-3.13 -- tests/azure-functions-durable/test_client_compat.py nox -R -s functions_e2e -- -k "dtask_client" ``` diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index ccaf9022..f2f59c94 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,8 +7,23 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +CHANGED + +- Dropped Python 3.10 support. Python 3.11 or later is now required. +- With the corresponding core SDK update, client and worker context managers +preserve Azure Managed subclass types in type checkers. +- With the corresponding core SDK update, multiple task completion callback +failures raise the built-in `ExceptionGroup`; a single failure is still raised +directly. The first callback error remains the explicit cause and serialized +inner failure. Branches on the previous aggregate exception type or message +can change during replay; see the [core upgrade warning](../CHANGELOG.md#unreleased). + FIXED +- With the corresponding core SDK update, grouped failures retain nested +exception messages, tracebacks, and notes in failure details. Tracebacks are +limited to 15 children per group and 10 group levels with truncation markers; +group children are not serialized as a complete structured tree. - With the corresponding core `durabletask` SDK fix, oversized messages fail large-payload validation before any uploads or payload replacements. Size errors now take precedence over hypothetical earlier storage failures. diff --git a/durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py b/durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py index 271f1bf8..95d219d5 100644 --- a/durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py +++ b/durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py @@ -2,7 +2,7 @@ # Licensed under the MIT License. import asyncio import os -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from threading import Lock from azure.core.credentials import AccessToken, TokenCredential @@ -62,14 +62,14 @@ def get_access_token(self) -> AccessToken | None: def is_token_expired(self) -> bool: if self.expiry_time is None: return True - return datetime.now(timezone.utc) >= (self.expiry_time - timedelta(seconds=self._refresh_interval_seconds)) + return datetime.now(UTC) >= (self.expiry_time - timedelta(seconds=self._refresh_interval_seconds)) def refresh_token(self): if self._credential is not None: self._token = self._credential.get_token(self._scope) # Convert UNIX timestamp to timezone-aware datetime - self.expiry_time = datetime.fromtimestamp(self._token.expires_on, tz=timezone.utc) + self.expiry_time = datetime.fromtimestamp(self._token.expires_on, tz=UTC) self._logger.debug(f"Token refreshed. Expires at: {self.expiry_time}") @@ -126,7 +126,7 @@ async def get_access_token(self) -> AccessToken | None: def is_token_expired(self) -> bool: if self.expiry_time is None: return True - return datetime.now(timezone.utc) >= ( + return datetime.now(UTC) >= ( self.expiry_time - timedelta(seconds=self._refresh_interval_seconds)) async def refresh_token(self): @@ -134,5 +134,5 @@ async def refresh_token(self): self._token = await self._credential.get_token(self._scope) # Convert UNIX timestamp to timezone-aware datetime - self.expiry_time = datetime.fromtimestamp(self._token.expires_on, tz=timezone.utc) + self.expiry_time = datetime.fromtimestamp(self._token.expires_on, tz=UTC) self._logger.debug(f"Token refreshed. Expires at: {self.expiry_time}") diff --git a/durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py b/durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py index c65c274f..1e7ce29e 100644 --- a/durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py +++ b/durabletask-azuremanaged/durabletask/azuremanaged/preview/sandboxes/client.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -from typing import Optional, Sequence +from typing import Optional, Self, Sequence import grpc from azure.core.credentials import TokenCredential @@ -48,7 +48,7 @@ def __init__( def close(self) -> None: self._transport.close() - def __enter__(self) -> "SandboxActivitiesClient": + def __enter__(self) -> Self: return self def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None: diff --git a/durabletask-azuremanaged/pyproject.toml b/durabletask-azuremanaged/pyproject.toml index 14e2f4c0..4e72a0c8 100644 --- a/durabletask-azuremanaged/pyproject.toml +++ b/durabletask-azuremanaged/pyproject.toml @@ -20,7 +20,6 @@ keywords = [ classifiers = [ "Development Status :: 5 - Production/Stable", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", @@ -28,7 +27,7 @@ classifiers = [ "Programming Language :: Python :: 3.15", "License :: OSI Approved :: MIT License", ] -requires-python = ">=3.10" +requires-python = ">=3.11" license = {file = "LICENSE"} readme = "README.md" dependencies = [ diff --git a/durabletask/client.py b/durabletask/client.py index b4dae185..d5269463 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -10,7 +10,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum -from typing import Any, Generic, Protocol, TypeVar, cast, overload +from typing import Any, Generic, Protocol, Self, TypeVar, cast, overload import grpc import grpc.aio @@ -621,7 +621,7 @@ def close(self) -> None: retired_channel.close() current_channel.close() - def __enter__(self) -> "TaskHubGrpcClient": + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: @@ -1106,7 +1106,7 @@ async def close(self) -> None: if channel is not None: await channel.close() - async def __aenter__(self) -> "AsyncTaskHubGrpcClient": + async def __aenter__(self) -> Self: return self async def __aexit__(self, *args: object) -> None: diff --git a/durabletask/entities/entity_instance_id.py b/durabletask/entities/entity_instance_id.py index e195736c..c1ea7e92 100644 --- a/durabletask/entities/entity_instance_id.py +++ b/durabletask/entities/entity_instance_id.py @@ -1,6 +1,8 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. +from typing import Self + class EntityInstanceId: __slots__ = ("_entity", "_key") @@ -29,13 +31,13 @@ def __delattr__(self, name: str) -> None: f"'{type(self).__name__}' object does not support attribute deletion" ) - def __copy__(self) -> "EntityInstanceId": + def __copy__(self) -> Self: return self - def __deepcopy__(self, memo: object) -> "EntityInstanceId": + def __deepcopy__(self, memo: object) -> Self: return self - def __reduce__(self) -> tuple[type["EntityInstanceId"], tuple[str, str]]: + def __reduce__(self) -> tuple[type[Self], tuple[str, str]]: return (type(self), (self._entity, self._key)) def __str__(self) -> str: diff --git a/durabletask/entities/entity_lock.py b/durabletask/entities/entity_lock.py index 87598964..226fa26c 100644 --- a/durabletask/entities/entity_lock.py +++ b/durabletask/entities/entity_lock.py @@ -3,7 +3,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Self if TYPE_CHECKING: @@ -15,7 +15,7 @@ class EntityLock: def __init__(self, context: OrchestrationContext): self._context = context - def __enter__(self) -> EntityLock: + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: diff --git a/durabletask/entities/entity_metadata.py b/durabletask/entities/entity_metadata.py index 91e329eb..7a3a517d 100644 --- a/durabletask/entities/entity_metadata.py +++ b/durabletask/entities/entity_metadata.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, TypeVar, overload from durabletask.entities.entity_instance_id import EntityInstanceId @@ -68,7 +68,7 @@ def from_entity_metadata(entity: pb.EntityMetadata, includes_state: bool, entity_state = entity.serializedState.value return EntityMetadata( id=entity_id, - last_modified=entity.lastModifiedTime.ToDatetime(timezone.utc), + last_modified=entity.lastModifiedTime.ToDatetime(UTC), backlog_queue_size=entity.backlogQueueSize, locked_by=entity.lockedBy.value, includes_state=includes_state, diff --git a/durabletask/extensions/azure_blob_payloads/blob_payload_store.py b/durabletask/extensions/azure_blob_payloads/blob_payload_store.py index bde74383..668bb758 100644 --- a/durabletask/extensions/azure_blob_payloads/blob_payload_store.py +++ b/durabletask/extensions/azure_blob_payloads/blob_payload_store.py @@ -9,6 +9,7 @@ import gzip import logging import uuid +from typing import Self from azure.core.exceptions import ResourceExistsError from azure.storage.blob import BlobServiceClient, ContainerClient @@ -119,13 +120,13 @@ async def close_async(self) -> None: """Close the underlying async blob service client.""" await self._async_blob_service_client.close() - def __enter__(self) -> BlobPayloadStore: + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: self.close() - async def __aenter__(self) -> BlobPayloadStore: + async def __aenter__(self) -> Self: return self async def __aexit__(self, *args: object) -> None: diff --git a/durabletask/extensions/history_export/_internal.py b/durabletask/extensions/history_export/_internal.py index 7b734702..0434e266 100644 --- a/durabletask/extensions/history_export/_internal.py +++ b/durabletask/extensions/history_export/_internal.py @@ -11,7 +11,7 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import UTC, datetime def dt_to_iso(value: datetime | None) -> str | None: @@ -19,9 +19,9 @@ def dt_to_iso(value: datetime | None) -> str | None: if value is None: return None if value.tzinfo is None: - value = value.replace(tzinfo=timezone.utc) + value = value.replace(tzinfo=UTC) else: - value = value.astimezone(timezone.utc) + value = value.astimezone(UTC) return value.isoformat() @@ -31,7 +31,7 @@ def dt_from_iso(value: str | None) -> datetime | None: return None parsed = datetime.fromisoformat(value) if parsed.tzinfo is None: - parsed = parsed.replace(tzinfo=timezone.utc) + parsed = parsed.replace(tzinfo=UTC) return parsed diff --git a/durabletask/extensions/history_export/activities.py b/durabletask/extensions/history_export/activities.py index b119afae..a3cd3eed 100644 --- a/durabletask/extensions/history_export/activities.py +++ b/durabletask/extensions/history_export/activities.py @@ -37,7 +37,7 @@ import hashlib from collections.abc import Callable, Mapping from dataclasses import dataclass -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, cast from durabletask import client as client_module @@ -288,7 +288,7 @@ def _dotnet_o_format(dt: datetime) -> str: a trailing zero. Naive datetimes are assumed UTC. """ if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) + dt = dt.replace(tzinfo=UTC) base = dt.strftime("%Y-%m-%dT%H:%M:%S") fractional = f"{dt.microsecond:06d}0" offset = dt.utcoffset() diff --git a/durabletask/extensions/history_export/azure_blob.py b/durabletask/extensions/history_export/azure_blob.py index 5919e6fc..48f744b4 100644 --- a/durabletask/extensions/history_export/azure_blob.py +++ b/durabletask/extensions/history_export/azure_blob.py @@ -19,7 +19,7 @@ from collections.abc import Mapping from dataclasses import dataclass, field -from typing import Any +from typing import Any, Self try: from azure.core.exceptions import ResourceExistsError @@ -117,7 +117,7 @@ def __init__(self, options: AzureBlobHistoryExportWriterOptions) -> None: def close(self) -> None: self._service.close() - def __enter__(self) -> "AzureBlobHistoryExportWriter": + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: diff --git a/durabletask/extensions/history_export/client.py b/durabletask/extensions/history_export/client.py index 1ab0ad54..abbd1eb2 100644 --- a/durabletask/extensions/history_export/client.py +++ b/durabletask/extensions/history_export/client.py @@ -42,8 +42,8 @@ desc = export_client.create_job(ExportJobCreationOptions( mode=ExportMode.BATCH, - completed_time_from=datetime(2026, 1, 1, tzinfo=timezone.utc), - completed_time_to=datetime(2026, 2, 1, tzinfo=timezone.utc), + completed_time_from=datetime(2026, 1, 1, tzinfo=UTC), + completed_time_to=datetime(2026, 2, 1, tzinfo=UTC), destination=ExportDestination(container="exports", prefix="january"), )) job_client = export_client.get_job_client(desc.job_id) @@ -56,7 +56,7 @@ import time import uuid from collections.abc import AsyncIterator, Iterator -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, cast import grpc @@ -180,7 +180,7 @@ def create_job( 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) + created_at = datetime.now(UTC) config_dict = config.to_dict() # A single ``create`` signal is enough: the entity validates @@ -502,7 +502,7 @@ async def create_job( 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) + created_at = datetime.now(UTC) await self._client.signal_entity( entity_id, ExportJobEntity.OP_CREATE, diff --git a/durabletask/extensions/history_export/entity.py b/durabletask/extensions/history_export/entity.py index 6afab6cd..27980718 100644 --- a/durabletask/extensions/history_export/entity.py +++ b/durabletask/extensions/history_export/entity.py @@ -41,7 +41,7 @@ from __future__ import annotations from collections.abc import Mapping -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, cast from durabletask import entities @@ -76,7 +76,7 @@ def _utcnow() -> datetime: - return datetime.now(timezone.utc) + return datetime.now(UTC) def _summarize_failures(failures: list[ExportFailure], *, limit: int = 10) -> str: diff --git a/durabletask/history.py b/durabletask/history.py index d3122d1c..c99c75b1 100644 --- a/durabletask/history.py +++ b/durabletask/history.py @@ -6,7 +6,7 @@ import functools from collections.abc import Callable from dataclasses import asdict, dataclass, fields -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any, cast from google.protobuf import json_format @@ -260,7 +260,7 @@ def to_dict(event: HistoryEvent) -> dict[str, Any]: def _base_kwargs(event: pb.HistoryEvent) -> dict[str, Any]: return { 'event_id': event.eventId, - 'timestamp': event.timestamp.ToDatetime(timezone.utc), + 'timestamp': event.timestamp.ToDatetime(UTC), } @@ -272,7 +272,7 @@ def _string_value(msg: Message, field_name: str) -> str | None: def _timestamp_value(msg: Message, field_name: str) -> datetime | None: if msg.HasField(field_name): - return getattr(msg, field_name).ToDatetime(timezone.utc) + return getattr(msg, field_name).ToDatetime(UTC) return None @@ -510,11 +510,11 @@ def _to_serializable(value: Any) -> Any: ), 'timerCreated': lambda event: TimerCreatedEvent( **_base_kwargs(event), - fire_at=event.timerCreated.fireAt.ToDatetime(timezone.utc), + fire_at=event.timerCreated.fireAt.ToDatetime(UTC), ), 'timerFired': lambda event: TimerFiredEvent( **_base_kwargs(event), - fire_at=event.timerFired.fireAt.ToDatetime(timezone.utc), + fire_at=event.timerFired.fireAt.ToDatetime(UTC), timer_id=event.timerFired.timerId, ), 'orchestratorStarted': lambda event: OrchestratorStartedEvent(**_base_kwargs(event)), diff --git a/durabletask/internal/client_helpers.py b/durabletask/internal/client_helpers.py index 06913c18..fc4fd7c9 100644 --- a/durabletask/internal/client_helpers.py +++ b/durabletask/internal/client_helpers.py @@ -6,7 +6,7 @@ import logging import uuid from collections.abc import Sequence -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any, TypeVar from google.protobuf import duration_pb2, wrappers_pb2 @@ -228,5 +228,5 @@ def build_signal_entity_req( requestId=str(uuid.uuid4()), scheduledTime=scheduled_time, parentTraceContext=None, - requestTime=helpers.new_timestamp(datetime.now(timezone.utc)) + requestTime=helpers.new_timestamp(datetime.now(UTC)) ) diff --git a/durabletask/internal/helpers.py b/durabletask/internal/helpers.py index 6e0b1a0a..c29e14d0 100644 --- a/durabletask/internal/helpers.py +++ b/durabletask/internal/helpers.py @@ -5,7 +5,7 @@ import traceback from collections.abc import Mapping, Sequence from decimal import Decimal -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, cast from google.protobuf import struct_pb2, timestamp_pb2, wrappers_pb2 @@ -254,12 +254,20 @@ def new_failure_details( get_qualified_name(type(ex)), exc_info=True) + if isinstance(ex, BaseExceptionGroup): + ex = cast(BaseExceptionGroup[BaseException], ex) + # Bound group rendering; the protobuf does not model every child. + stack_trace = traceback.TracebackException.from_exception( + ex, max_group_width=15, max_group_depth=10).format() + elif isinstance(ex, Exception): + stack_trace = traceback.format_tb(ex.__traceback__) + else: + stack_trace = traceback.format_exception(ex) + failure_details = pb.TaskFailureDetails( errorType=get_qualified_name(type(ex)), errorMessage=str(ex), - stackTrace=wrappers_pb2.StringValue(value=''.join( - traceback.format_tb(ex.__traceback__) if isinstance(ex, Exception) - else traceback.format_exception(type(ex), ex, ex.__traceback__))), + stackTrace=wrappers_pb2.StringValue(value=''.join(stack_trace)), innerFailure=( new_failure_details( inner, exception_properties_provider, logger, _visited) @@ -547,7 +555,7 @@ def ensure_aware(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: - return value.replace(tzinfo=timezone.utc) + return value.replace(tzinfo=UTC) return value diff --git a/durabletask/scheduled/models.py b/durabletask/scheduled/models.py index e8927a69..75f10f69 100644 --- a/durabletask/scheduled/models.py +++ b/durabletask/scheduled/models.py @@ -33,12 +33,6 @@ def _to_iso(value: datetime | None) -> str | None: def _from_iso(value: str | None) -> datetime | None: if not value: return None - # .NET serializes ``DateTimeOffset`` with a numeric offset, but tolerate a - # trailing ``Z`` so states written by other producers still parse on the - # Python versions that predate ``fromisoformat`` accepting ``Z``. Only the - # trailing designator is normalized so an interior ``Z`` is left untouched. - if value.endswith("Z"): - value = value[:-1] + "+00:00" return datetime.fromisoformat(value) diff --git a/durabletask/scheduled/schedule_entity.py b/durabletask/scheduled/schedule_entity.py index 90f385e7..dab5ce5b 100644 --- a/durabletask/scheduled/schedule_entity.py +++ b/durabletask/scheduled/schedule_entity.py @@ -2,7 +2,7 @@ # Licensed under the MIT License. import logging -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any from durabletask.entities import DurableEntity, EntityInstanceId @@ -24,7 +24,7 @@ def _now() -> datetime: - return datetime.now(timezone.utc) + return datetime.now(UTC) class Schedule(DurableEntity): diff --git a/durabletask/serialization.py b/durabletask/serialization.py index 22d09b22..45d045a0 100644 --- a/durabletask/serialization.py +++ b/durabletask/serialization.py @@ -39,7 +39,6 @@ import json import logging import re -import sys import types import typing from abc import ABC, abstractmethod @@ -681,7 +680,6 @@ def _build_dataclass(cls: Any, data: dict[str, Any], hints = typing.get_type_hints(cls) except Exception: hints = {} - globalns = _type_namespace(cls) kwargs: dict[str, Any] = {} for field in dataclasses.fields(cls): if field.name not in data: @@ -702,62 +700,6 @@ def _build_dataclass(cls: Any, data: dict[str, Any], # Bound classmethods omit it from their inspected signature; # staticmethods have no implicit receiver. continue - # ``get_type_hints`` on Python 3.10 does not deep-resolve forward - # references nested inside container args (e.g. the ``"TreeNode"`` in - # ``list["TreeNode"]`` on a self-referential dataclass), leaving a bare - # string or ``ForwardRef`` that the coercion below would skip. Resolve - # them against the class's defining module so reconstruction behaves the - # same as it does on 3.11+. - field_type = _resolve_forward_refs(hints.get(field.name), globalns) + field_type = hints.get(field.name) kwargs[field.name] = _coerce_to_type(data[field.name], field_type, converter) return cls(**kwargs) - - -def _type_namespace(cls: Any) -> dict[str, Any]: - """Build the namespace used to resolve forward references in ``cls``'s hints. - - Forward references in a class's annotations are resolved against the - module in which the class is defined, plus the class's own name (so a - self-referential type like ``list["TreeNode"]`` resolves). - """ - module = sys.modules.get(getattr(cls, "__module__", None) or "") - ns: dict[str, Any] = dict(getattr(module, "__dict__", {})) - name = getattr(cls, "__name__", None) - if name: - ns.setdefault(name, cls) - return ns - - -def _resolve_forward_refs(tp: Any, globalns: dict[str, Any]) -> Any: - """Resolve string / ``ForwardRef`` leaves in a type hint, recursing into args. - - Returns ``tp`` unchanged when it (or a nested name) cannot be resolved, so an - unresolvable hint simply falls back to "leave the value as parsed JSON" - rather than raising. Only the supported generic shapes (``Union``, ``list``, - ``dict``, ``tuple``, etc.) are rebuilt; the destination type is still - entirely caller-supplied, so this does not weaken the security model. - """ - if isinstance(tp, str): - try: - tp = eval(tp, globalns) - except Exception: - return tp - elif isinstance(tp, typing.ForwardRef): - try: - tp = eval(tp.__forward_arg__, globalns) - except Exception: - return tp - - origin = typing.get_origin(tp) - if origin is None: - return tp - args = typing.get_args(tp) - if not args: - return tp - resolved = [_resolve_forward_refs(a, globalns) for a in args] - if origin is typing.Union or origin is types.UnionType: - return typing.Union[tuple(resolved)] - try: - return origin[tuple(resolved) if len(resolved) > 1 else resolved[0]] - except TypeError: - return tp diff --git a/durabletask/task.py b/durabletask/task.py index b2ca7b32..17cc7f3d 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -9,8 +9,8 @@ from abc import ABC, abstractmethod from collections.abc import Callable, Generator, Sequence from dataclasses import dataclass -from datetime import datetime, timedelta, timezone -from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast, overload +from datetime import UTC, datetime, timedelta +from typing import Any, Generic, TypeAlias, TypeVar, cast, overload from durabletask.entities import DurableEntity, EntityInstanceId, EntityLock, EntityContext import durabletask.internal.helpers as pbh @@ -450,16 +450,7 @@ def create_replay_safe_logger(self, logger: logging.Logger) -> ReplaySafeLogger: return ReplaySafeLogger(logger, lambda: self.is_replaying) -if TYPE_CHECKING: - # logging.LoggerAdapter is generic in stubs but is not subscriptable - # at runtime before Python 3.11. Use a TYPE_CHECKING alias so the - # base class evaluates correctly at runtime. - _LoggerAdapterBase = logging.LoggerAdapter[logging.Logger] -else: - _LoggerAdapterBase = logging.LoggerAdapter - - -class ReplaySafeLogger(_LoggerAdapterBase): +class ReplaySafeLogger(logging.LoggerAdapter[logging.Logger]): """A logger adapter that suppresses log messages during orchestration replay. This class extends :class:`logging.LoggerAdapter` and only emits log @@ -608,16 +599,6 @@ class TaskCancelledError(Exception): """Exception type for cancelled orchestration tasks.""" -class _TaskCompletionError(Exception): - """Collect callback errors without requiring Python 3.11 ExceptionGroup.""" - - def __init__(self, errors: list[Exception]) -> None: - self.errors = tuple(errors) - super().__init__( - "Multiple task completion callbacks failed: " - + "; ".join(f"{type(error).__name__}: {error}" for error in errors)) - - class Task(ABC, Generic[T]): """Abstract base class for asynchronous tasks in a durable orchestration.""" _result: T @@ -673,7 +654,7 @@ def _notify_parent(self) -> None: if len(errors) == 1: raise errors[0] if errors: - raise _TaskCompletionError(errors) from errors[0] + raise ExceptionGroup("Multiple task completion callbacks failed", errors) from errors[0] class CompositeTask(Task[T]): @@ -858,7 +839,7 @@ def compute_next_delay(self) -> timedelta | None: else: backoff_coefficient = self._retry_policy.backoff_coefficient - if datetime.now(tz=timezone.utc).replace(tzinfo=None) < retry_expiration: + if datetime.now(tz=UTC).replace(tzinfo=None) < retry_expiration: next_delay_f = math.pow(backoff_coefficient, self._attempt_count - 1) * self._retry_policy.first_retry_interval.total_seconds() if self._retry_policy.max_retry_interval is not None: diff --git a/durabletask/testing/in_memory_backend.py b/durabletask/testing/in_memory_backend.py index 44aea1f7..0ac60e05 100644 --- a/durabletask/testing/in_memory_backend.py +++ b/durabletask/testing/in_memory_backend.py @@ -18,7 +18,7 @@ from collections import deque from collections.abc import Callable, Iterable, Iterator from dataclasses import dataclass, field -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import TypeAlias, cast import grpc @@ -49,8 +49,8 @@ class OrchestrationInstance: input: str | None = None output: str | None = None custom_status: str | None = None - created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) - last_updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + last_updated_at: datetime = field(default_factory=lambda: datetime.now(UTC)) completed_at: datetime | None = None failure_details: pb.TaskFailureDetails | None = None history: list[pb.HistoryEvent] = field(default_factory=_new_history_event_list) @@ -76,7 +76,7 @@ class EntityState: """Internal entity state stored by the in-memory backend.""" instance_id: str serialized_state: str | None = None - last_modified_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + last_modified_at: datetime = field(default_factory=lambda: datetime.now(UTC)) locked_by: str | None = None pending_operations: list[pb.HistoryEvent] = field(default_factory=_new_history_event_list) dispatched_operations: list[pb.HistoryEvent] = field(default_factory=_new_history_event_list) @@ -254,8 +254,8 @@ def StartInstance(self, request: pb.CreateInstanceRequest, context: grpc.Service f"Orchestration instance '{instance_id}' already exists") return pb.CreateInstanceResponse(instanceId=instance_id) - now = datetime.now(timezone.utc) - start_time = request.scheduledStartTimestamp.ToDatetime(tzinfo=timezone.utc) \ + now = datetime.now(UTC) + start_time = request.scheduledStartTimestamp.ToDatetime(tzinfo=UTC) \ if request.HasField("scheduledStartTimestamp") else now version = request.version.value if request.HasField("version") else None @@ -350,7 +350,7 @@ def RaiseEvent(self, request: pb.RaiseEventRequest, context: grpc.ServicerContex request.input.value if request.input else None ) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) # Don't dispatch work for suspended or terminal orchestrations; # suspended events will be delivered when the orchestration is @@ -390,7 +390,7 @@ def SuspendInstance(self, request: pb.SuspendRequest, context: grpc.ServicerCont encoded_input=request.reason.value if request.HasField("reason") else None ) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(instance.instance_id) self._logger.info(f"Suspended instance '{request.instanceId}'") @@ -409,7 +409,7 @@ def ResumeInstance(self, request: pb.ResumeRequest, context: grpc.ServicerContex encoded_input=request.reason.value if request.HasField("reason") else None ) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(instance.instance_id) self._logger.info(f"Resumed instance '{request.instanceId}'") @@ -437,9 +437,9 @@ def PurgeInstances(self, request: pb.PurgeInstancesRequest, context: grpc.Servic continue if pf.runtimeStatus and inst.status not in pf.runtimeStatus: continue - if pf.HasField("createdTimeFrom") and inst.created_at < pf.createdTimeFrom.ToDatetime(timezone.utc): + if pf.HasField("createdTimeFrom") and inst.created_at < pf.createdTimeFrom.ToDatetime(UTC): continue - if pf.HasField("createdTimeTo") and inst.created_at >= pf.createdTimeTo.ToDatetime(timezone.utc): + if pf.HasField("createdTimeTo") and inst.created_at >= pf.createdTimeTo.ToDatetime(UTC): continue to_purge.append(iid) for iid in to_purge: @@ -500,9 +500,9 @@ def ListInstanceIds(self, request: pb.ListInstanceIdsRequest, context: grpc.Serv continue if instance.completed_at is None: continue - if request.HasField("completedTimeFrom") and instance.completed_at < request.completedTimeFrom.ToDatetime(timezone.utc): + if request.HasField("completedTimeFrom") and instance.completed_at < request.completedTimeFrom.ToDatetime(UTC): continue - if request.HasField("completedTimeTo") and instance.completed_at >= request.completedTimeTo.ToDatetime(timezone.utc): + if request.HasField("completedTimeTo") and instance.completed_at >= request.completedTimeTo.ToDatetime(UTC): continue matching.append(instance) @@ -515,7 +515,7 @@ def ListInstanceIds(self, request: pb.ListInstanceIdsRequest, context: grpc.Serv if request.HasField("lastInstanceKey") and request.lastInstanceKey.value: token = request.lastInstanceKey.value sep_idx = token.index(_TOKEN_SEP) - token_ts = datetime.fromisoformat(token[:sep_idx]).replace(tzinfo=timezone.utc) + token_ts = datetime.fromisoformat(token[:sep_idx]).replace(tzinfo=UTC) token_id = token[sep_idx + 1:] # bisect_right positions us just after the cursor entry start_index = bisect.bisect_right(sort_keys, (token_ts, token_id)) @@ -623,7 +623,7 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context: grpc.ServicerCo if instance.history and not ( instance.dispatched_events and instance.dispatched_events[0].HasField("orchestratorStarted")): - now = datetime.now(timezone.utc) + now = datetime.now(UTC) orch_started = helpers.new_orchestrator_started_event(now) instance.dispatched_events.insert(0, orch_started) @@ -772,7 +772,7 @@ def CompleteOrchestratorTask(self, request: pb.OrchestratorResponse, context: gr new_events = list(instance.dispatched_events) instance.history.extend(new_events) instance.dispatched_events.clear() - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) if request.customStatus: instance.custom_status = request.customStatus.value @@ -877,7 +877,7 @@ def CompleteActivityTask(self, request: pb.ActivityResponse, context: grpc.Servi ) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(request.instanceId) return pb.CompleteTaskResponse() @@ -903,7 +903,7 @@ def CompleteEntityTask(self, request: pb.EntityBatchResult, context: grpc.Servic entity.serialized_state = request.entityState.value else: entity.serialized_state = None - entity.last_modified_at = datetime.now(timezone.utc) + entity.last_modified_at = datetime.now(UTC) entity.dispatched_operations.clear() # Update completion token for next batch @@ -946,7 +946,7 @@ def CompleteEntityTask(self, request: pb.EntityBatchResult, context: grpc.Servic continue parent_instance.pending_events.append(event) - parent_instance.last_updated_at = datetime.now(timezone.utc) + parent_instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(parent_instance_id) except Exception: self._logger.exception( @@ -959,7 +959,7 @@ def CompleteEntityTask(self, request: pb.EntityBatchResult, context: grpc.Servic if action.HasField("sendSignal"): signal = action.sendSignal scheduled_time = ( - signal.scheduledTime.ToDatetime(tzinfo=timezone.utc) + signal.scheduledTime.ToDatetime(tzinfo=UTC) if signal.HasField("scheduledTime") else None ) self._signal_entity_internal( @@ -990,7 +990,7 @@ def CompleteEntityTask(self, request: pb.EntityBatchResult, context: grpc.Servic def SignalEntity(self, request: pb.SignalEntityRequest, context: grpc.ServicerContext) -> pb.SignalEntityResponse: """Signals an entity, queueing an operation for processing.""" scheduled_time = ( - request.scheduledTime.ToDatetime(tzinfo=timezone.utc) + request.scheduledTime.ToDatetime(tzinfo=UTC) if request.HasField("scheduledTime") else None ) with self._lock: @@ -1015,7 +1015,7 @@ def SignalEntity(self, request: pb.SignalEntityRequest, context: grpc.ServicerCo targetInstanceId=wrappers_pb2.StringValue(value=entity_id), ) ) - if scheduled_time is not None and scheduled_time > datetime.now(timezone.utc): + if scheduled_time is not None and scheduled_time > datetime.now(UTC): self._schedule_delayed_entity_operation(entity_id, event, scheduled_time) else: entity.pending_operations.append(event) @@ -1062,9 +1062,9 @@ def QueryInstances(self, request: pb.QueryInstancesRequest, context: grpc.Servic if query.runtimeStatus and instance.status not in query.runtimeStatus: continue # Filter by created time range - if query.HasField("createdTimeFrom") and instance.created_at < query.createdTimeFrom.ToDatetime(timezone.utc): + if query.HasField("createdTimeFrom") and instance.created_at < query.createdTimeFrom.ToDatetime(UTC): continue - if query.HasField("createdTimeTo") and instance.created_at >= query.createdTimeTo.ToDatetime(timezone.utc): + if query.HasField("createdTimeTo") and instance.created_at >= query.createdTimeTo.ToDatetime(UTC): continue # Filter by instance ID prefix if query.HasField("instanceIdPrefix") and query.instanceIdPrefix.value: @@ -1130,9 +1130,9 @@ def QueryEntities(self, request: pb.QueryEntitiesRequest, context: grpc.Servicer if not entity.instance_id.startswith(query.instanceIdStartsWith.value): continue # Filter by last modified time range - if query.HasField("lastModifiedFrom") and entity.last_modified_at < query.lastModifiedFrom.ToDatetime(timezone.utc): + if query.HasField("lastModifiedFrom") and entity.last_modified_at < query.lastModifiedFrom.ToDatetime(UTC): continue - if query.HasField("lastModifiedTo") and entity.last_modified_at >= query.lastModifiedTo.ToDatetime(timezone.utc): + if query.HasField("lastModifiedTo") and entity.last_modified_at >= query.lastModifiedTo.ToDatetime(UTC): continue # Filter transient (entities with pending operations) if not query.includeTransient and entity.pending_operations: @@ -1321,7 +1321,7 @@ def _create_instance_internal(self, instance_id: str, name: str, else: raise ValueError(f"Orchestration instance '{instance_id}' already exists") - now = datetime.now(timezone.utc) + now = datetime.now(UTC) instance = OrchestrationInstance( instance_id=instance_id, name=name, @@ -1355,7 +1355,7 @@ def _raise_event_internal(self, instance_id: str, event_name: str, event = helpers.new_event_raised_event(event_name, event_data) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(instance.instance_id) def _terminate_instance_internal(self, instance_id: str, output: str | None, @@ -1374,7 +1374,7 @@ def _terminate_instance_internal(self, instance_id: str, output: str | None, event = helpers.new_terminated_event(encoded_output=output) instance.pending_events.append(event) - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(instance.instance_id) self._logger.info(f"Terminated instance '{instance_id}'") @@ -1486,7 +1486,7 @@ def _process_complete_orchestration_action(self, instance: OrchestrationInstance if complete_action.HasField("failureDetails") else None ) - instance.completed_at = datetime.now(timezone.utc) if self._is_terminal_status(status) else None + instance.completed_at = datetime.now(UTC) if self._is_terminal_status(status) else None if status == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW: # Handle continue-as-new @@ -1513,7 +1513,7 @@ def _process_complete_orchestration_action(self, instance: OrchestrationInstance # Build the new pending events in the correct order: # OrchestratorStarted, ExecutionStarted, carryover, new arrivals - now = datetime.now(timezone.utc) + now = datetime.now(UTC) execution_started = helpers.new_execution_started_event( instance.name, instance.instance_id, new_input, version=new_version, @@ -1566,7 +1566,7 @@ def _process_create_timer_action(self, instance: OrchestrationInstance, """Processes a create timer action.""" create_timer = action.createTimer timer_id = action.id - fire_at = create_timer.fireAt.ToDatetime(tzinfo=timezone.utc) + fire_at = create_timer.fireAt.ToDatetime(tzinfo=UTC) # Add TimerCreated event to history timer_created_event = helpers.new_timer_created_event(timer_id, fire_at) @@ -1577,7 +1577,7 @@ def _process_create_timer_action(self, instance: OrchestrationInstance, instance.status = pb.ORCHESTRATION_STATUS_RUNNING # Schedule timer firing - now = datetime.now(timezone.utc) + now = datetime.now(UTC) delay = max(0, (fire_at - now).total_seconds()) def fire_timer() -> None: @@ -1587,7 +1587,7 @@ def fire_timer() -> None: if current_instance and not self._is_terminal_status(current_instance.status): timer_fired_event = helpers.new_timer_fired_event(timer_id, fire_at) current_instance.pending_events.append(timer_fired_event) - current_instance.last_updated_at = datetime.now(timezone.utc) + current_instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(instance.instance_id) timer_thread = threading.Thread(target=fire_timer, daemon=True) @@ -1655,7 +1655,7 @@ def watch() -> None: event = helpers.new_sub_orchestration_failed_event(task_id, Exception(error_msg)) parent_instance.pending_events.append(event) - parent_instance.last_updated_at = datetime.now(timezone.utc) + parent_instance.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(parent_instance_id) watcher_thread = threading.Thread(target=watch, daemon=True) @@ -1718,10 +1718,10 @@ def _process_send_entity_message_action(self, instance: OrchestrationInstance, entityOperationSignaled=signaled, ) scheduled_time = ( - signaled.scheduledTime.ToDatetime(tzinfo=timezone.utc) + signaled.scheduledTime.ToDatetime(tzinfo=UTC) if signaled.HasField("scheduledTime") else None ) - if scheduled_time is not None and scheduled_time > datetime.now(timezone.utc): + if scheduled_time is not None and scheduled_time > datetime.now(UTC): self._schedule_delayed_entity_operation(target_id, operation_event, scheduled_time) else: self._queue_entity_operation(target_id, operation_event) @@ -1828,7 +1828,7 @@ def _grant_lock(self, pending: PendingLockRequest): ), ) parent.pending_events.append(grant_event) - parent.last_updated_at = datetime.now(timezone.utc) + parent.last_updated_at = datetime.now(UTC) self._enqueue_orchestration(pending.parent_instance_id) def _try_grant_lock(self, pending: PendingLockRequest) -> bool: @@ -1881,7 +1881,7 @@ def _signal_entity_internal(self, entity_id: str, operation: str, targetInstanceId=wrappers_pb2.StringValue(value=entity_id), ) ) - if scheduled_time is not None and scheduled_time > datetime.now(timezone.utc): + if scheduled_time is not None and scheduled_time > datetime.now(UTC): self._schedule_delayed_entity_operation(entity_id, event, scheduled_time) else: self._queue_entity_operation(entity_id, event) @@ -1894,7 +1894,7 @@ def _schedule_delayed_entity_operation(self, entity_id: str, event: pb.HistoryEv for orchestration timers, so that delayed entity signals are honored by the in-memory backend. """ - delay = max(0.0, (fire_at - datetime.now(timezone.utc)).total_seconds()) + delay = max(0.0, (fire_at - datetime.now(UTC)).total_seconds()) # Capture the lifecycle generation so a timer outliving a stop()/reset() # does not enqueue into a restarted or cleared backend. scheduled_generation = self._generation @@ -1929,7 +1929,7 @@ def _prepare_rewind(self, instance: OrchestrationInstance, instance.status = pb.ORCHESTRATION_STATUS_RUNNING instance.output = None instance.failure_details = None - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) # Clear any stale dispatched events. instance.dispatched_events.clear() @@ -1939,7 +1939,7 @@ def _prepare_rewind(self, instance: OrchestrationInstance, # audit event carries a meaningful timestamp, consistent with # other backend-generated events. rewind_timestamp = timestamp_pb2.Timestamp() - rewind_timestamp.FromDatetime(datetime.now(timezone.utc)) + rewind_timestamp.FromDatetime(datetime.now(UTC)) rewind_event = pb.HistoryEvent( eventId=-1, timestamp=rewind_timestamp, @@ -1973,7 +1973,7 @@ def _process_rewind_orchestration_action( instance.status = pb.ORCHESTRATION_STATUS_RUNNING instance.output = None instance.failure_details = None - instance.last_updated_at = datetime.now(timezone.utc) + instance.last_updated_at = datetime.now(UTC) # Identify sub-orchestrations that were created but did not # complete successfully — they need to be recursively rewound. @@ -2026,7 +2026,7 @@ def _process_rewind_orchestration_action( instance.pending_events.clear() instance.dispatched_events.clear() instance.pending_events.append( - helpers.new_orchestrator_started_event(datetime.now(timezone.utc))) + helpers.new_orchestrator_started_event(datetime.now(UTC))) instance.completion_token = self._next_completion_token self._next_completion_token += 1 self._orchestration_in_flight.discard(instance.instance_id) diff --git a/durabletask/worker.py b/durabletask/worker.py index df601552..1d4b59f8 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -7,17 +7,16 @@ import json import logging import os -import sys import time from collections.abc import Callable, Generator, Sequence from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from threading import Event, Lock, Thread from types import GeneratorType from enum import Enum from functools import partial -from typing import Any, TypeVar, cast, overload +from typing import Any, Self, TypeVar, cast, overload import uuid from packaging.version import InvalidVersion, parse @@ -64,11 +63,11 @@ def _is_control_flow_exception(error: BaseException) -> bool: - if isinstance(error, (asyncio.CancelledError, GeneratorExit, KeyboardInterrupt, SystemExit)): - return True - if sys.version_info >= (3, 11) and isinstance(error, BaseExceptionGroup): - return any(_is_control_flow_exception(inner) for inner in error.exceptions) - return False + control_flow_types = (asyncio.CancelledError, GeneratorExit, KeyboardInterrupt, SystemExit) + if isinstance(error, BaseExceptionGroup): + group = cast(BaseExceptionGroup[BaseException], error) + return group.subgroup(control_flow_types) is not None + return isinstance(error, control_flow_types) class ConcurrencyOptions: @@ -628,7 +627,7 @@ def emit_trace_spans(self) -> bool: """Return whether this worker emits Durable Task lifecycle spans.""" return self._emit_trace_spans - def __enter__(self) -> "TaskHubGrpcWorker": + def __enter__(self) -> Self: return self def __exit__(self, *args: object) -> None: @@ -980,7 +979,7 @@ def stream_reader(response_stream: Any = response_stream) -> None: loop.run_in_executor(None, work_item_queue.get), timeout=queue_timeout, ) - except asyncio.TimeoutError: + except TimeoutError: work_item = None stream_outcome = self._classify_stream_outcome( saw_message=saw_message, @@ -1497,7 +1496,7 @@ def _execute_entity_batch( entity_method_cache=self._entity_method_cache, ) for operation in req.operations: - start_time = datetime.now(timezone.utc) + start_time = datetime.now(UTC) operation_result = None @@ -1525,7 +1524,7 @@ def _execute_entity_batch( operation_result = pb.OperationResult(success=pb.OperationResultSuccess( result=entity_result, startTimeUtc=new_timestamp(start_time), - endTimeUtc=new_timestamp(datetime.now(timezone.utc)) + endTimeUtc=new_timestamp(datetime.now(UTC)) )) results.append(operation_result) @@ -1537,7 +1536,7 @@ def _execute_entity_batch( failureDetails=ph.new_failure_details( ex, self._exception_properties_provider, self._logger), startTimeUtc=new_timestamp(start_time), - endTimeUtc=new_timestamp(datetime.now(timezone.utc)) + endTimeUtc=new_timestamp(datetime.now(UTC)) )) results.append(operation_result) @@ -1840,7 +1839,7 @@ def create_timer_internal( # tzinfo whose utcoffset() is None is still naive and must be left as-is # (calling astimezone() on it would raise ValueError). if final_fire_at.utcoffset() is not None: - final_fire_at = final_fire_at.astimezone(timezone.utc).replace(tzinfo=None) + final_fire_at = final_fire_at.astimezone(UTC).replace(tzinfo=None) next_fire_at: datetime = final_fire_at @@ -3869,7 +3868,7 @@ async def _consume_queue(self, queue: asyncio.Queue[_WorkItem], semaphore: async try: try: work = await asyncio.wait_for(queue.get(), timeout=1.0) - except asyncio.TimeoutError: + except TimeoutError: continue func, cancellation_func, args, kwargs = work diff --git a/examples/custom_data_converter/src/workflows.py b/examples/custom_data_converter/src/workflows.py index 61e8bb0d..0c2a4152 100644 --- a/examples/custom_data_converter/src/workflows.py +++ b/examples/custom_data_converter/src/workflows.py @@ -14,7 +14,7 @@ from __future__ import annotations from collections.abc import Generator -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any from pydantic import BaseModel, Field, field_validator @@ -111,7 +111,7 @@ def sample_order() -> Order: """A valid sample order used by both the runnable app and the tests.""" return Order( customer="Contoso", - placed_at=datetime.now(timezone.utc), + placed_at=datetime.now(UTC), items=[ OrderItem(name="Widget", quantity=3, unit_price=25.0), OrderItem(name="Gadget", quantity=1, unit_price=99.99), diff --git a/examples/distributed-tracing/README.md b/examples/distributed-tracing/README.md index d47a045e..e0375ba3 100644 --- a/examples/distributed-tracing/README.md +++ b/examples/distributed-tracing/README.md @@ -16,7 +16,7 @@ all produce correlated trace spans: ## Prerequisites - [Docker](https://www.docker.com/) (for the emulator and Jaeger) -- Python 3.10+ +- Python 3.11+ ## Quick Start diff --git a/examples/history_export/app.py b/examples/history_export/app.py index 6a786f0f..d7a9234e 100644 --- a/examples/history_export/app.py +++ b/examples/history_export/app.py @@ -20,7 +20,7 @@ import os import time from collections.abc import Generator -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from typing import Any from durabletask import client, task, worker @@ -99,7 +99,7 @@ def main() -> None: time.sleep(0.5) # Create an export job for the seeded window. - now = datetime.now(timezone.utc) + now = datetime.now(UTC) print("\nCreating export job...") desc = export_client.create_job( ExportJobCreationOptions( diff --git a/examples/large_payload/README.md b/examples/large_payload/README.md index 89104801..a1e3aa0e 100644 --- a/examples/large_payload/README.md +++ b/examples/large_payload/README.md @@ -20,7 +20,7 @@ your orchestrator or activity code. ## Prerequisites -- Python 3.10+ +- Python 3.11+ - [Docker](https://www.docker.com/) (for the DTS emulator) - [Azurite](https://learn.microsoft.com/azure/storage/common/storage-use-azurite) or an Azure Storage account diff --git a/examples/scheduled_tasks.py b/examples/scheduled_tasks.py index 1490889e..759a5414 100644 --- a/examples/scheduled_tasks.py +++ b/examples/scheduled_tasks.py @@ -9,7 +9,7 @@ import os import time from collections.abc import Generator -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from typing import Any from azure.identity import DefaultAzureCredential @@ -53,7 +53,7 @@ def greet_orchestrator(ctx: task.OrchestrationContext, name: str) -> Generator[t orchestration_name=task.get_name(greet_orchestrator), interval=timedelta(seconds=5), orchestration_input="world", - start_at=datetime.now(timezone.utc), + start_at=datetime.now(UTC), start_immediately_if_late=True, )) diff --git a/noxfile.py b/noxfile.py index 252b8d51..996f9730 100644 --- a/noxfile.py +++ b/noxfile.py @@ -5,7 +5,7 @@ Run ``nox -s ci`` for a representative local validation sweep. Use a focused session while developing (for example, ``nox -s lint`` or -``nox -s core_tests-3.10``). The test sessions provision their required +``nox -s core_tests-3.11``). The test sessions provision their required emulators when possible. Usage: @@ -13,8 +13,8 @@ nox -s lint nox -s typecheck_core nox -s typecheck_functions_package - nox -s core_tests-3.10 - nox -s azuremanaged_tests-3.10 + nox -s core_tests-3.11 + nox -s azuremanaged_tests-3.11 nox -s functions_unit-3.13 nox -s functions_e2e @@ -46,8 +46,8 @@ E2E_APPS_DIR = REPO_ROOT / "tests" / "azure-functions-durable" / "e2e" / "apps" # Sample apps that need an in-app virtual environment for the E2E suite. E2E_APPS = ("v1_style", "dtask_style", "tracing") -PYTHON_VERSIONS = ("3.10", "3.11", "3.12", "3.13", "3.14", "3.15") -DEFAULT_CI_PYTHON = "3.10" +PYTHON_VERSIONS = ("3.11", "3.12", "3.13", "3.14", "3.15") +DEFAULT_CI_PYTHON = "3.11" def _install_packages(session: nox.Session, editable: bool = False) -> None: @@ -294,7 +294,7 @@ def lint(session: nox.Session) -> None: session.run("flake8", *targets) -@nox.session(python=["3.10"]) +@nox.session(python=["3.11"]) def typecheck_core(session: nox.Session) -> None: """Run strict pyright checks for the core and Azure Managed packages.""" session.install("-r", "requirements.txt") diff --git a/pyproject.toml b/pyproject.toml index ebe594c3..4d3ca9f8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,6 @@ keywords = [ classifiers = [ "Development Status :: 5 - Production/Stable", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", @@ -27,13 +26,12 @@ classifiers = [ "Programming Language :: Python :: 3.15", "License :: OSI Approved :: MIT License", ] -requires-python = ">=3.10" +requires-python = ">=3.11" license = {file = "LICENSE"} readme = "README.md" dependencies = [ "grpcio", "protobuf", - "asyncio", "packaging" ] diff --git a/pyrightconfig.json b/pyrightconfig.json index 46a54dbd..0cef3aa9 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -2,7 +2,8 @@ "include": [ "durabletask", "durabletask-azuremanaged", - "examples" + "examples", + "tests/durabletask/typing" ], "exclude": [ "**/__pycache__", @@ -13,6 +14,6 @@ "**/.venv*", ".venv*" ], - "pythonVersion": "3.10", + "pythonVersion": "3.11", "typeCheckingMode": "strict" } diff --git a/requirements.txt b/requirements.txt index 98fac019..2e0a5d7e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,7 +8,6 @@ pytest-cov azure-identity azure-functions typing-extensions -asyncio packaging opentelemetry-api opentelemetry-sdk \ No newline at end of file diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index 1529b91c..ea85f0d1 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/tests/azure-functions-durable/test_worker_compat.py @@ -84,6 +84,31 @@ def orchestrator(context): assert len(response.actions) == 1 +@pytest.mark.parametrize("use_async", [False, True]) +async def test_worker_preserves_grouped_failure_diagnostics(monkeypatch, use_async): + monkeypatch.setattr(payloads, "_payload_store", None) + + def orchestrator(context): + error = ValueError("nested failure") + error.add_note("nested note") + raise ExceptionGroup("activity group", [error, RuntimeError("other failure")]) + + worker = DurableFunctionsWorker() + request = _encode_orchestrator_request("grouped-failure") + if use_async: + encoded = await worker.execute_orchestration_request_async(orchestrator, request) + else: + encoded = worker.execute_orchestration_request(orchestrator, request) + response = _decode_orchestrator_response(encoded) + completion = _get_completion_action(response) + + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.failureDetails.errorType == "builtins.ExceptionGroup" + assert "ValueError: nested failure" in completion.failureDetails.stackTrace.value + assert "RuntimeError: other failure" in completion.failureDetails.stackTrace.value + assert "nested note" in completion.failureDetails.stackTrace.value + + def test_worker_created_before_configuration_hydrates_and_externalizes(monkeypatch, payload_store_factory): monkeypatch.setattr(payloads, "_payload_store", None) worker = DurableFunctionsWorker() diff --git a/tests/azure-functions-durable/typing/import_package.py b/tests/azure-functions-durable/typing/import_package.py index e9395042..8d69f92d 100644 --- a/tests/azure-functions-durable/typing/import_package.py +++ b/tests/azure-functions-durable/typing/import_package.py @@ -2,8 +2,11 @@ # Licensed under the MIT License. from datetime import timedelta +from typing import assert_type import azure.durable_functions as df +from azure.durable_functions.client import DurableFunctionsClient, SyncDurableFunctionsClient +from azure.durable_functions.worker import DurableFunctionsWorker def create_app() -> df.DFApp: @@ -21,3 +24,18 @@ def create_retry_policy() -> df.RetryPolicy: def activity_is_complete(context: df.DurableOrchestrationContext) -> bool: return context.call_activity("activity").is_complete + + +def check_sync_context_managers( + client: SyncDurableFunctionsClient, + worker: DurableFunctionsWorker, +) -> None: + with client as entered_client: + assert_type(entered_client, SyncDurableFunctionsClient) + with worker as entered_worker: + assert_type(entered_worker, DurableFunctionsWorker) + + +async def check_async_context_manager(client: DurableFunctionsClient) -> None: + async with client as entered_client: + assert_type(entered_client, DurableFunctionsClient) diff --git a/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py b/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py index 0a96eaf9..d8038cb3 100644 --- a/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py +++ b/tests/durabletask-azuremanaged/test_azuremanaged_grpc_resiliency.py @@ -71,6 +71,32 @@ def orchestrator(ctx: task.OrchestrationContext, _: Any): assert response.actions[0].completeOrchestration.failureDetails.errorMessage == "deterministic teardown failure" +def test_dts_worker_preserves_grouped_failure_diagnostics(): + def orchestrator(ctx: task.OrchestrationContext, _: Any): + error = ValueError("nested failure") + error.add_note("nested note") + raise ExceptionGroup("activity group", [error, RuntimeError("other failure")]) + + with DurableTaskSchedulerWorker( + host_address="localhost:4001", taskhub="hub", token_credential=None) as worker: + name = worker.add_orchestrator(orchestrator) + request = pb.OrchestratorRequest(instanceId="instance", newEvents=[ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event(name, "instance"), + ]) + stub = Mock() + worker._execute_orchestrator(request, stub, "original-token") + + stub.CompleteOrchestratorTask.assert_called_once() + stub.AbandonTaskOrchestratorWorkItem.assert_not_called() + completion = stub.CompleteOrchestratorTask.call_args.args[0].actions[0].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.failureDetails.errorType == "builtins.ExceptionGroup" + assert "ValueError: nested failure" in completion.failureDetails.stackTrace.value + assert "RuntimeError: other failure" in completion.failureDetails.stackTrace.value + assert "nested note" in completion.failureDetails.stackTrace.value + + def test_dts_worker_passes_exception_properties_provider_to_base_worker(): class PropertiesProvider: def get_exception_properties(self, exception: Exception) -> Mapping[str, Any] | None: diff --git a/tests/durabletask/entities/test_entity_id_parsing.py b/tests/durabletask/entities/test_entity_id_parsing.py index 1a9fb942..91429e6a 100644 --- a/tests/durabletask/entities/test_entity_id_parsing.py +++ b/tests/durabletask/entities/test_entity_id_parsing.py @@ -158,6 +158,15 @@ def custom_name(self) -> str: return self.entity.upper() +def test_entity_instance_id_subclass_copy(): + custom = _SubclassEntityId("Counter", "key1") + assert copy.copy(custom) is custom + assert copy.deepcopy(custom) is custom + constructor, args = custom.__reduce__() + assert constructor is _SubclassEntityId + assert args == ("counter", "key1") + + def test_entity_instance_id_subclass_pickle(): """Verify that unpickling a subclass of EntityInstanceId preserves the concrete type.""" custom = _SubclassEntityId("Counter", "key1") diff --git a/tests/durabletask/scheduled/test_models.py b/tests/durabletask/scheduled/test_models.py index 410ff800..a655668d 100644 --- a/tests/durabletask/scheduled/test_models.py +++ b/tests/durabletask/scheduled/test_models.py @@ -45,6 +45,24 @@ def test_valid_options(self): class TestCreationOptionsSerialization: + @pytest.mark.parametrize("timestamp, expected", [ + ("2026-01-01T12:34:56Z", datetime(2026, 1, 1, 12, 34, 56, tzinfo=timezone.utc)), + ("2026-01-01T12:34:56+00:00", datetime(2026, 1, 1, 12, 34, 56, tzinfo=timezone.utc)), + ("2026-01-01T14:34:56+02:00", datetime(2026, 1, 1, 12, 34, 56, tzinfo=timezone.utc)), + ("2026-01-01T12:34:56", datetime(2026, 1, 1, 12, 34, 56)), + ]) + def test_parse_iso_timestamps(self, timestamp: str, expected: datetime) -> None: + options = ScheduleCreationOptions.from_json({ + "schedule_id": "s1", + "orchestration_name": "orch", + "interval_seconds": 5, + "start_at": timestamp, + "end_at": timestamp, + }) + assert options.start_at == options.end_at == expected + assert options.start_at is not None + assert options.start_at.isoformat() == timestamp.replace("Z", "+00:00") + def test_round_trip_through_json(self): start = datetime(2026, 1, 1, tzinfo=timezone.utc) end = datetime(2026, 2, 1, tzinfo=timezone.utc) diff --git a/tests/durabletask/test_data_converter.py b/tests/durabletask/test_data_converter.py index 1ed5c552..e954e9d5 100644 --- a/tests/durabletask/test_data_converter.py +++ b/tests/durabletask/test_data_converter.py @@ -471,6 +471,32 @@ def activity(ctx: Any, values: Any) -> Any: assert default.deserialize(payload, target_type) == json.loads(payload) +@pytest.mark.parametrize("option", ["enable_generic_sets", "enable_set_coercion"]) +def test_activity_set_forward_references_use_native_type_hint_resolution(option: str): + def activity( + ctx: Any, values: set[tuple["PricedOrder", ...]], + ) -> frozenset[tuple["PricedOrder", ...]]: + assert values == {(PricedOrder(3, 10),)} + return frozenset(values) + + converter = JsonDataConverter(**{option: True}) + input_type = type_discovery.activity_input_type(activity, converter) + output_type = type_discovery.activity_output_type(activity, converter) + assert input_type == set[tuple[PricedOrder, ...]] + assert output_type == frozenset[tuple[PricedOrder, ...]] + assert type_discovery.activity_input_type(activity) is None + assert type_discovery.activity_output_type(activity) is None + + registry = _Registry() + name = registry.add_activity(activity) + executor = _ActivityExecutor(registry, logging.getLogger("test_set_forward_refs"), converter) + payload = '[[{"quantity": 3, "unit_price": 10, "total": 999}]]' + output = executor.execute("instance", name, 1, payload) + restored = converter.deserialize(output, output_type) + assert isinstance(restored, frozenset) + assert restored == frozenset({(PricedOrder(3, 10),)}) + + @pytest.mark.parametrize("target_type", [ set[tuple[list[int], int]], set[tuple[dict[str, int], ...]], diff --git a/tests/durabletask/test_failure_details.py b/tests/durabletask/test_failure_details.py index 7700837e..539c5347 100644 --- a/tests/durabletask/test_failure_details.py +++ b/tests/durabletask/test_failure_details.py @@ -4,6 +4,7 @@ """Unit tests for :class:`durabletask.task.FailureDetails` introspection.""" import logging +import traceback from decimal import Decimal from typing import cast @@ -65,6 +66,100 @@ def test_new_failure_details_uses_fqn_for_custom_type(): assert details.errorType == helpers.get_qualified_name(_DerivedError) +def test_new_failure_details_preserves_ordinary_exception_stack_format(): + try: + raise ValueError("ordinary failure") + except ValueError as error: + details = helpers.new_failure_details(error) + assert details.stackTrace.value == "".join(traceback.format_tb(error.__traceback__)) + assert details.errorMessage == "ordinary failure" + + +@pytest.mark.parametrize("base_group", [False, True]) +def test_new_failure_details_preserves_grouped_tracebacks_and_notes(base_group: bool): + class UnexpectedExit(BaseException): + pass + + try: + raise ValueError("nested failure") + except ValueError as error: + error.add_note("nested diagnostic note") + nested = ExceptionGroup("inner group", [error]) + other = UnexpectedExit("unexpected exit") if base_group else RuntimeError("other failure") + group = BaseExceptionGroup("outer group", [nested, other]) + group.add_note("group diagnostic note") + try: + raise group + except BaseExceptionGroup as caught: + details = helpers.new_failure_details(caught) + assert details.stackTrace.value == "".join(traceback.format_exception(caught)) + + assert details.errorType == ("builtins.BaseExceptionGroup" if base_group else "builtins.ExceptionGroup") + assert details.errorMessage == str(group) + assert "ValueError: nested failure" in details.stackTrace.value + assert "nested diagnostic note" in details.stackTrace.value + assert "group diagnostic note" in details.stackTrace.value + state = client.parse_orchestration_state(pb.OrchestrationState(failureDetails=details)) + assert state.failure_details is not None + assert state.failure_details.stack_trace == details.stackTrace.value + + +@pytest.mark.parametrize("base_group", [False, True]) +@pytest.mark.parametrize("child_count", [15, 20]) +def test_grouped_failure_diagnostics_limit_width(base_group: bool, child_count: int): + errors: list[BaseException] = [ValueError(f"child-{index:02}") for index in range(child_count)] + if base_group: + errors[0] = BaseException("child-00") + for index, error in enumerate(errors): + error.add_note(f"note-{index:02}") + group = BaseExceptionGroup("wide group", errors) + group.add_note("group note") + details = helpers.new_failure_details(group) + stack = details.stackTrace.value + + assert len(group.exceptions) == child_count + assert "group note" in stack + for index in range(15): + assert f"child-{index:02}" in stack + assert f"note-{index:02}" in stack + if child_count > 15: + assert "and 5 more exceptions" in stack + for index in range(15, child_count): + assert f"child-{index:02}" not in stack + assert f"note-{index:02}" not in stack + else: + assert "more exceptions" not in stack + assert not details.HasField("innerFailure") + assert helpers.failure_details_from_protobuf(details).stack_trace == stack + + +@pytest.mark.parametrize("base_group", [False, True]) +@pytest.mark.parametrize("group_depth", [10, 11, 13]) +def test_grouped_failure_diagnostics_limit_depth(base_group: bool, group_depth: int): + error: BaseException = BaseException("deep leaf") if base_group else ValueError("deep leaf") + error.add_note("leaf note") + for depth in range(group_depth - 1, 0, -1): + error = BaseExceptionGroup(f"level-{depth:02}", [error]) + group = BaseExceptionGroup("deep group", [error]) + group.add_note("group note") + details = helpers.new_failure_details(group) + stack = details.stackTrace.value + + assert "group note" in stack + assert "level-09" in stack + if group_depth > 10: + assert "max_group_depth is 10" in stack + assert "level-10" not in stack + assert "deep leaf" not in stack + assert "leaf note" not in stack + else: + assert "max_group_depth" not in stack + assert "deep leaf" in stack + assert "leaf note" in stack + assert not details.HasField("innerFailure") + assert helpers.failure_details_from_protobuf(details).stack_trace == stack + + # --------------------------------------------------------------------------- # is_caused_by with an exception type (base-type aware) # --------------------------------------------------------------------------- diff --git a/tests/durabletask/test_orchestration_dispatch_recovery.py b/tests/durabletask/test_orchestration_dispatch_recovery.py index 4d4f7ee6..7d6527a8 100644 --- a/tests/durabletask/test_orchestration_dispatch_recovery.py +++ b/tests/durabletask/test_orchestration_dispatch_recovery.py @@ -6,7 +6,6 @@ import asyncio import json import logging -import sys from threading import Event from typing import Any, Generator from unittest.mock import Mock @@ -430,22 +429,26 @@ def orchestrator(ctx: task.OrchestrationContext, _: Any): def _control_flow_errors() -> list[BaseException]: - errors: list[BaseException] = [ + return [ asyncio.CancelledError("cancel"), SystemExit("exit"), KeyboardInterrupt("interrupt"), GeneratorExit("close"), + BaseExceptionGroup("grouped cancellation", [asyncio.CancelledError("cancel")]), + BaseExceptionGroup("mixed", [ValueError("ordinary"), asyncio.CancelledError("cancel")]), + BaseExceptionGroup("nested", [ + TeardownExit("unexpected"), + BaseExceptionGroup("inner", [SystemExit("exit")]), + ]), + BaseExceptionGroup("nested interrupt", [ + ExceptionGroup("ordinary", [ValueError("ordinary")]), + BaseExceptionGroup("inner", [KeyboardInterrupt("interrupt")]), + ]), + BaseExceptionGroup("nested generator exit", [ + BaseExceptionGroup("inner", [GeneratorExit("close")]), + TeardownExit("unexpected"), + ]), ] - if sys.version_info >= (3, 11): - errors.extend([ - BaseExceptionGroup("grouped cancellation", [asyncio.CancelledError("cancel")]), - BaseExceptionGroup("mixed", [ValueError("ordinary"), asyncio.CancelledError("cancel")]), - BaseExceptionGroup("nested", [ - TeardownExit("unexpected"), - BaseExceptionGroup("inner", [SystemExit("exit")]), - ]), - ]) - return errors @pytest.mark.parametrize("error", _control_flow_errors(), ids=str) @@ -591,20 +594,59 @@ def test_failure_details_do_not_pass_base_exception_to_public_properties_provide provider.get_exception_properties.assert_not_called() -@pytest.mark.skipif(sys.version_info < (3, 11), reason="Exception groups require Python 3.11+.") def test_grouped_unexpected_failure_keeps_nested_reasons(): - if sys.version_info >= (3, 11): - def orchestrator(ctx: task.OrchestrationContext, _: Any): - try: - yield ctx.wait_for_external_event("ready") - finally: - raise BaseExceptionGroup("teardown", [TeardownExit("root cause"), ValueError("second cause")]) - - result = _executor(orchestrator).execute(INSTANCE_ID, [], _request().newEvents) - failure = result.actions[0].completeOrchestration.failureDetails - assert failure.errorType.endswith(".BaseExceptionGroup") - assert "TeardownExit: root cause" in failure.stackTrace.value - assert "ValueError: second cause" in failure.stackTrace.value + def orchestrator(ctx: task.OrchestrationContext, _: Any): + try: + yield ctx.wait_for_external_event("ready") + finally: + raise BaseExceptionGroup("teardown", [TeardownExit("root cause"), ValueError("second cause")]) + + result = _executor(orchestrator).execute(INSTANCE_ID, [], _request().newEvents) + failure = result.actions[0].completeOrchestration.failureDetails + assert failure.errorType.endswith(".BaseExceptionGroup") + assert "TeardownExit: root cause" in failure.stackTrace.value + assert "ValueError: second cause" in failure.stackTrace.value + + +@pytest.mark.parametrize("base_group", [False, True]) +@pytest.mark.parametrize("during_teardown", [False, True]) +def test_grouped_failure_preserves_dispatch_outcome_and_diagnostics( + base_group: bool, during_teardown: bool): + second = TeardownExit("unexpected exit") if base_group else RuntimeError("ordinary failure") + nested = ValueError("nested reason") + nested.add_note("nested note") + error = BaseExceptionGroup("grouped failure", [ExceptionGroup("inner", [nested]), second]) + + def orchestrator(ctx: task.OrchestrationContext, _: Any): + if not during_teardown: + raise error + try: + yield ctx.call_activity("work") + finally: + raise error + + with worker.TaskHubGrpcWorker(logger=LOGGER) as host: + host._registry = _executor(orchestrator)._registry + stub = Mock() + reset = Mock() + host._execute_orchestrator( + _request(), stub, "original-token", _on_outcome_uncertain=reset) + + stub.CompleteOrchestratorTask.assert_called_once() + stub.AbandonTaskOrchestratorWorkItem.assert_not_called() + reset.assert_not_called() + response = stub.CompleteOrchestratorTask.call_args.args[0] + if during_teardown and not base_group: + assert len(response.actions) == 1 + assert response.actions[0].scheduleTask.name == "work" + assert not response.actions[0].HasField("completeOrchestration") + return + completion = response.actions[-1].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.failureDetails.errorType == helpers.get_qualified_name(type(error)) + assert "ValueError: nested reason" in completion.failureDetails.stackTrace.value + assert "nested note" in completion.failureDetails.stackTrace.value + assert f"{type(second).__name__}: {second}" in completion.failureDetails.stackTrace.value @pytest.mark.asyncio diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 80f96f78..0d3d7840 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -613,6 +613,42 @@ def dummy_activity(ctx, _): assert not hasattr(activity_task, "is_cancelled") +def test_cancel_callback_group_replays_legacy_first_cause_recovery(): + def orchestrator(ctx: task.OrchestrationContext, _): + timeout = ctx.create_timer(timedelta(hours=1)) + task.when_all([timeout]) + task.when_all([timeout]) + yield ctx.wait_for_external_event("cancel") + try: + timeout.cancel() + except Exception as error: + recovery = ( + "cancellation-recovery" + if isinstance(error.__cause__, task.TaskCancelledError) + else "other-recovery") + return (yield ctx.call_activity(recovery)) + return None + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + start_time = datetime(2026, 1, 1) + # The previous callback wrapper chained the first TaskCancelledError, so + # this history records the cancellation branch, not the aggregate's type. + history = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + helpers.new_timer_created_event(1, start_time + timedelta(hours=1)), + helpers.new_event_raised_event("cancel"), + helpers.new_task_scheduled_event(2, "cancellation-recovery"), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute( + TEST_INSTANCE_ID, history, [helpers.new_task_completed_event(2, '"recovered"')]) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completion.result.value == '"recovered"' + + def test_cancelled_task_get_result_raises_task_cancelled_error(): """Tests that cancelled cancellable tasks raise TaskCancelledError from get_result.""" diff --git a/tests/durabletask/test_serialization.py b/tests/durabletask/test_serialization.py index 2e70d2eb..e771c179 100644 --- a/tests/durabletask/test_serialization.py +++ b/tests/durabletask/test_serialization.py @@ -16,7 +16,7 @@ from collections import namedtuple from dataclasses import dataclass from types import SimpleNamespace -from typing import Any, List, Optional, Union, get_args +from typing import Any, FrozenSet, List, Optional, Set, Union import pytest from durabletask import serialization @@ -25,7 +25,6 @@ from durabletask.serialization import _encode_custom_object from durabletask.serialization import _coerce_to_type as coerce_to_type from durabletask.serialization import _from_json as from_json -from durabletask.serialization import _resolve_forward_refs as resolve_forward_refs from durabletask.serialization import _to_json as to_json # ----- Test fixtures ----- @@ -541,42 +540,29 @@ def from_json(cls, value, strict=False): assert result == Flagged("ok") -# ----- forward-reference resolution (Python 3.10 get_type_hints parity) ----- -# -# On Python 3.10, ``typing.get_type_hints`` does not deep-resolve forward -# references nested inside container args (e.g. the element type of -# ``list["TreeNode"]`` on a self-referential dataclass), leaving a bare string -# or ``ForwardRef``. ``_resolve_forward_refs`` restores the 3.11+ behavior so -# nested coercion still fires. These tests run on every supported version. - - -def test_resolve_forward_refs_resolves_string_element_type(): - # ``list["Address"]`` evaluates to a generic whose arg is the raw string - # "Address" -- exactly what 3.10 leaves behind. - resolved = resolve_forward_refs(List["Address"], {"Address": Address}) - assert get_args(resolved) == (Address,) - - -def test_resolve_forward_refs_resolves_forwardref_element_type(): - resolved = resolve_forward_refs(List["Address"], {"Address": Address}) - assert get_args(resolved) == (Address,) - +def test_dataclass_nested_forward_references_round_trip(): + @dataclass + class Container: + items: list["Address"] + legacy_items: List["Address"] + by_name: dict[str, "Address"] + optional: Optional["Address"] + pair: tuple["Address", ...] -def test_resolve_forward_refs_then_coerce_reconstructs_nested_dataclass(): - # The full 3.10 path: resolve the unresolved element type, then coerce the - # contained dicts into the target dataclass. - field_type = resolve_forward_refs(list["Address"], {"Address": Address}) - result = coerce_to_type([{"street": "a", "city": "b"}], field_type) - assert isinstance(result[0], Address) - assert result[0].city == "b" + address = Address("a", "b") + value = Container([address], [address], {"home": address}, address, (address,)) + assert from_json(to_json(value), Container) == value -def test_resolve_forward_refs_leaves_unresolvable_name_untouched(): - # An unknown name is left as a string so coercion harmlessly falls back to - # the parsed JSON rather than raising. - resolved = resolve_forward_refs(list["DoesNotExist"], {}) - assert get_args(resolved) == ("DoesNotExist",) +def test_dataclass_unresolvable_annotation_preserves_parsed_json(): + @dataclass + class Container: + value: Any + Container.__annotations__["value"] = "DoesNotExist" + result = from_json('{"value": {"street": "a", "city": "b"}}', Container) + assert isinstance(result, Container) + assert result.value == {"street": "a", "city": "b"} # ----- set and frozenset container serialization and coercion ----- @@ -587,6 +573,55 @@ class FrozenAddress: city: str +@dataclass(frozen=True) +class FrozenTreeNode: + name: str + children: frozenset["FrozenTreeNode"] = frozenset() + + +@pytest.mark.parametrize("option", ["enable_generic_sets", "enable_set_coercion"]) +def test_dataclass_nested_set_forward_references_round_trip(option: str): + @dataclass + class Container: + items: set["FrozenAddress"] + legacy_items: Set["FrozenAddress"] + frozen_items: frozenset["FrozenAddress"] + legacy_frozen_items: FrozenSet["FrozenAddress"] + nested: dict[str, list[frozenset[tuple["FrozenAddress", ...]]]] + optional: frozenset["FrozenAddress"] | None + + address = FrozenAddress("a", "b") + value = Container( + {address}, {address}, frozenset({address}), frozenset({address}), + {"home": [frozenset({(address,)})]}, frozenset({address}), + ) + converter = JsonDataConverter(**{option: True}) + encoded = converter.serialize(value) + restored = converter.deserialize(encoded, Container) + assert restored == value + assert isinstance(restored.items, set) + assert isinstance(restored.legacy_items, set) + assert isinstance(restored.frozen_items, frozenset) + assert isinstance(restored.legacy_frozen_items, frozenset) + assert isinstance(restored.optional, frozenset) + assert converter.serialize(restored) == encoded + + +@pytest.mark.parametrize("option", ["enable_generic_sets", "enable_set_coercion"]) +def test_dataclass_self_referential_frozenset_round_trip(option: str): + value = FrozenTreeNode("root", frozenset({ + FrozenTreeNode("branch", frozenset({FrozenTreeNode("leaf")})), + FrozenTreeNode("other"), + })) + converter = JsonDataConverter(**{option: True}) + encoded = converter.serialize(value) + restored = converter.deserialize(encoded, FrozenTreeNode) + assert restored == value + assert isinstance(restored.children, frozenset) + assert all(isinstance(child, FrozenTreeNode) for child in restored.children) + assert converter.serialize(restored) == encoded + + class SetItem: """Identity-hashable value exposing a custom JSON representation.""" diff --git a/tests/durabletask/test_task_composition.py b/tests/durabletask/test_task_composition.py index 168126dc..898a37ea 100644 --- a/tests/durabletask/test_task_composition.py +++ b/tests/durabletask/test_task_composition.py @@ -11,6 +11,7 @@ import pytest from durabletask import task +from durabletask.internal import helpers from durabletask.internal import orchestrator_service_pb2 as pb @@ -172,7 +173,7 @@ def test_shared_retryable_terminal_completion(failed: bool) -> None: assert root.get_result() == [["retried", "retried"], ["retried"]] -@pytest.mark.parametrize("group_count", [1, 2]) +@pytest.mark.parametrize("group_count", [1, 2, 20]) def test_cancel_callback_errors_do_not_block_other_observers(group_count: int) -> None: child = task.CancellableTask[str]() broken = [task.when_all([child, child]) for _ in range(group_count)] @@ -186,10 +187,15 @@ def test_cancel_callback_errors_do_not_block_other_observers(group_count: int) - if group_count == 1: assert isinstance(raised.value, task.TaskCancelledError) else: - assert isinstance(raised.value, task._TaskCompletionError) - assert len(raised.value.errors) == group_count + assert isinstance(raised.value, ExceptionGroup) + assert len(raised.value.exceptions) == group_count assert all(isinstance(error, task.TaskCancelledError) - for error in raised.value.errors) + for error in raised.value.exceptions) + assert raised.value.__cause__ is raised.value.exceptions[0] + details = helpers.failure_details_from_protobuf(helpers.new_failure_details(raised.value)) + assert details.inner_failure is not None + assert details.inner_failure.is_caused_by(task.TaskCancelledError) + assert details.inner_failure.message == str(raised.value.__cause__) assert child.is_cancelled assert not child.cancel() assert race.get_result() is child @@ -199,6 +205,45 @@ def test_cancel_callback_errors_do_not_block_other_observers(group_count: int) - assert not child._parents +def test_callback_exception_group_preserves_errors_and_tracebacks() -> None: + class ThrowingGroup(task.WhenAllTask[str]): + def __init__(self, child: task.Task[str], error: Exception) -> None: + self.error = error + super().__init__([child]) + + def on_child_completed(self, child: task.Task[Any]) -> None: + super().on_child_completed(child) + raise self.error + + child = task.CompletableTask[str]() + first_error = ValueError("first callback") + second_error = RuntimeError("second callback") + first = ThrowingGroup(child, first_error) + second = ThrowingGroup(child, second_error) + healthy = task.when_all([child]) + + with pytest.raises(ExceptionGroup) as raised: + child.complete("ok") + + assert raised.value.exceptions == (first_error, second_error) + assert raised.value.__cause__ is first_error + assert all(error.__traceback__ is not None for error in raised.value.exceptions) + value_errors, other_errors = raised.value.split(ValueError) + assert value_errors is not None + assert value_errors.exceptions == (first_error,) + assert other_errors is not None + assert other_errors.exceptions == (second_error,) + details = helpers.new_failure_details(raised.value) + assert "ValueError: first callback" in details.stackTrace.value + assert "RuntimeError: second callback" in details.stackTrace.value + assert details.HasField("innerFailure") + assert details.innerFailure.errorType == "builtins.ValueError" + assert details.innerFailure.errorMessage == "first callback" + assert details.innerFailure.stackTrace.value == helpers.new_failure_details(first_error).stackTrace.value + assert first.get_result() == second.get_result() == healthy.get_result() == ["ok"] + assert not child._parents + + def test_throwing_cancel_handler_leaves_task_pending() -> None: child = task.CancellableTask[str]() left = task.when_all([child]) diff --git a/tests/durabletask/typing/context_managers.py b/tests/durabletask/typing/context_managers.py new file mode 100644 index 00000000..dc05d88e --- /dev/null +++ b/tests/durabletask/typing/context_managers.py @@ -0,0 +1,74 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Static regression checks for subclass-preserving instance return types.""" + +from typing import assert_type + +from durabletask.azuremanaged.client import ( + AsyncDurableTaskSchedulerClient, + DurableTaskSchedulerClient, +) +from durabletask.azuremanaged.preview.sandboxes import SandboxActivitiesClient +from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker +from durabletask.entities import EntityInstanceId, EntityLock +from durabletask.extensions.azure_blob_payloads import BlobPayloadStore +from durabletask.extensions.history_export.azure_blob import AzureBlobHistoryExportWriter + + +class CustomEntityLock(EntityLock): + pass + + +class CustomEntityId(EntityInstanceId): + pass + + +class CustomPayloadStore(BlobPayloadStore): + pass + + +class CustomHistoryWriter(AzureBlobHistoryExportWriter): + pass + + +class CustomSandboxClient(SandboxActivitiesClient): + pass + + +def check_sync_context_managers( + client: DurableTaskSchedulerClient, + worker: DurableTaskSchedulerWorker, + entity_lock: CustomEntityLock, + store: CustomPayloadStore, + writer: CustomHistoryWriter, + sandbox: CustomSandboxClient, +) -> None: + with client as entered_client: + assert_type(entered_client, DurableTaskSchedulerClient) + with worker as entered_worker: + assert_type(entered_worker, DurableTaskSchedulerWorker) + with entity_lock as entered_lock: + assert_type(entered_lock, CustomEntityLock) + with store as entered_store: + assert_type(entered_store, CustomPayloadStore) + with writer as entered_writer: + assert_type(entered_writer, CustomHistoryWriter) + with sandbox as entered_sandbox: + assert_type(entered_sandbox, CustomSandboxClient) + + +async def check_async_context_managers( + client: AsyncDurableTaskSchedulerClient, + store: CustomPayloadStore, +) -> None: + async with client as entered_client: + assert_type(entered_client, AsyncDurableTaskSchedulerClient) + async with store as entered_store: + assert_type(entered_store, CustomPayloadStore) + + +def check_entity_copy_types(entity_id: CustomEntityId) -> None: + assert_type(entity_id.__copy__(), CustomEntityId) + assert_type(entity_id.__deepcopy__({}), CustomEntityId) + assert_type(entity_id.__reduce__(), tuple[type[CustomEntityId], tuple[str, str]])