From af7019bfc291841d147e408e625c651e83d843c3 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Tue, 20 Oct 2026 10:24:32 -0700 Subject: [PATCH] Drop Python 3.10 support and adopt Python 3.11 features Raise the core and Azure Managed minimum to Python 3.11 while retaining Python 3.13 for Functions. Align CI, Nox, type checking, and documentation; use Self, UTC and ExceptionGroup, and remove obsolete compatibility shims. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e6b07f3f-b3e1-49dd-9121-318968272d31 --- .github/copilot-instructions.md | 3 +- .../workflows/durabletask-azuremanaged.yml | 2 +- .github/workflows/durabletask.yml | 2 +- .github/workflows/typecheck.yml | 6 +- CHANGELOG.md | 9 ++ README.md | 5 ++ azure-functions-durable/CHANGELOG.md | 6 ++ .../azure/durable_functions/http/builtin.py | 6 +- .../compat/durable_orchestration_status.py | 4 +- .../internal/compat/history_projection.py | 4 +- .../internal/compat/orchestration_context.py | 4 +- .../internal/history_export_compat.py | 4 +- .../azure/durable_functions/testing/entity.py | 4 +- docs/development.md | 8 +- durabletask-azuremanaged/CHANGELOG.md | 9 ++ .../internal/access_token_manager.py | 10 +-- .../azuremanaged/preview/sandboxes/client.py | 4 +- durabletask-azuremanaged/pyproject.toml | 3 +- durabletask/client.py | 6 +- durabletask/entities/entity_lock.py | 4 +- durabletask/entities/entity_metadata.py | 4 +- .../azure_blob_payloads/blob_payload_store.py | 5 +- .../extensions/history_export/_internal.py | 8 +- .../extensions/history_export/activities.py | 4 +- .../extensions/history_export/azure_blob.py | 4 +- .../extensions/history_export/client.py | 10 +-- .../extensions/history_export/entity.py | 4 +- durabletask/history.py | 10 +-- durabletask/internal/client_helpers.py | 4 +- durabletask/internal/helpers.py | 4 +- durabletask/scheduled/models.py | 6 -- durabletask/scheduled/schedule_entity.py | 4 +- durabletask/serialization.py | 60 +------------ durabletask/task.py | 29 ++---- durabletask/testing/in_memory_backend.py | 88 +++++++++---------- durabletask/worker.py | 18 ++-- .../custom_data_converter/src/workflows.py | 4 +- examples/distributed-tracing/README.md | 2 +- examples/history_export/app.py | 4 +- examples/large_payload/README.md | 2 +- examples/scheduled_tasks.py | 4 +- noxfile.py | 12 +-- pyproject.toml | 3 +- pyrightconfig.json | 5 +- .../typing/import_package.py | 18 ++++ tests/durabletask/scheduled/test_models.py | 18 ++++ tests/durabletask/test_serialization.py | 53 +++++------ tests/durabletask/test_task_composition.py | 37 +++++++- tests/durabletask/typing/context_managers.py | 64 ++++++++++++++ 49 files changed, 328 insertions(+), 263 deletions(-) create mode 100644 tests/durabletask/typing/context_managers.py 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 4cef9ed4..b3910301 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 16194e95..e11217a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ 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. +- Context managers now preserve subclass types in type checkers, including +Azure Managed clients and workers. +- Multiple task completion callback failures now raise the built-in +`ExceptionGroup`, preserving each exception and its traceback. A single callback +failure is still raised directly. + ## v1.12.0 ADDED 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 40f767b6..32f7b384 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -18,6 +18,12 @@ 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. ## v2.0.0rc3 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 338b183f..709ab60c 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,15 @@ 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. + ## v1.12.0 ADDED 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 4bec371a..24dc30d8 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 @@ -620,7 +620,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: @@ -1105,7 +1105,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_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 cd277429..565c31a5 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 @@ -227,5 +227,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 0e170371..a406cb45 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 @@ -545,7 +545,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 52c3dfed..d13932ab 100644 --- a/durabletask/serialization.py +++ b/durabletask/serialization.py @@ -38,7 +38,6 @@ import inspect import json import logging -import sys import types import typing from abc import ABC, abstractmethod @@ -537,7 +536,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: @@ -558,62 +556,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) # noqa: S307 - resolves caller-authored annotations - except Exception: - return tp - elif isinstance(tp, typing.ForwardRef): - try: - tp = eval(tp.__forward_arg__, globalns) # noqa: S307 - 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..0677a0ee 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) 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 25010a17..0e2a221f 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -11,11 +11,11 @@ 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 typing import Any, TypeVar, cast, overload +from typing import Any, Self, TypeVar, cast, overload import uuid from packaging.version import InvalidVersion, parse @@ -618,7 +618,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: @@ -974,7 +974,7 @@ def stream_reader() -> 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, @@ -1418,7 +1418,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 @@ -1446,7 +1446,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) @@ -1458,7 +1458,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) @@ -1751,7 +1751,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 @@ -3704,7 +3704,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..4b8bf917 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,7 +26,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/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/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/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_serialization.py b/tests/durabletask/test_serialization.py index 02861dcd..63bd524c 100644 --- a/tests/durabletask/test_serialization.py +++ b/tests/durabletask/test_serialization.py @@ -12,14 +12,13 @@ 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, List, Optional, Union import pytest from durabletask.serialization import _AUTO_SERIALIZED as AUTO_SERIALIZED 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 @@ -536,38 +535,26 @@ 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", ...] + 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_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" +def test_dataclass_unresolvable_annotation_preserves_parsed_json(): + @dataclass + class Container: + value: Any -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",) + Container.__annotations__["value"] = "DoesNotExist" + result = from_json('{"value": {"street": "a", "city": "b"}}', Container) + assert isinstance(result, Container) + assert result.value == {"street": "a", "city": "b"} diff --git a/tests/durabletask/test_task_composition.py b/tests/durabletask/test_task_composition.py index 168126dc..bb508cb4 100644 --- a/tests/durabletask/test_task_composition.py +++ b/tests/durabletask/test_task_composition.py @@ -186,10 +186,10 @@ 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 child.is_cancelled assert not child.cancel() assert race.get_result() is child @@ -199,6 +199,37 @@ 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 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,) + 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..239f6d2a --- /dev/null +++ b/tests/durabletask/typing/context_managers.py @@ -0,0 +1,64 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Static regression checks for inherited context-manager 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 EntityLock +from durabletask.extensions.azure_blob_payloads import BlobPayloadStore +from durabletask.extensions.history_export.azure_blob import AzureBlobHistoryExportWriter + + +class CustomEntityLock(EntityLock): + 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)