Fix linter errors and update documentation

Author: cgillum

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added `set_custom_status` orchestrator API ([#31](https://github.com/microsoft/durabletask-python/pull/31)) - contributed by [@famarting](https://github.com/famarting)
 - Added `purge_orchestration` client API ([#34](https://github.com/microsoft/durabletask-python/pull/34)) - contributed by [@famarting](https://github.com/famarting)
+- Added new `durabletask-azuremanaged` package for use with the [Durable Task Scheduler](https://techcommunity.microsoft.com/blog/appsonazureblog/announcing-limited-early-access-of-the-durable-task-scheduler-for-azure-durable-/4286526) - by [@RyanLettieri](https://github.com/RyanLettieri)
 
 ### Changes
 

README.md

@@ -1,15 +1,14 @@
-# Durable Task Client SDK for Python
+# Durable Task SDK for Python
 
 [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
 [![Build Validation](https://github.com/microsoft/durabletask-python/actions/workflows/pr-validation.yml/badge.svg)](https://github.com/microsoft/durabletask-python/actions/workflows/pr-validation.yml)
 [![PyPI version](https://badge.fury.io/py/durabletask.svg)](https://badge.fury.io/py/durabletask)
 
-This repo contains a Python client SDK for use with the [Durable Task Framework for Go](https://github.com/microsoft/durabletask-go) and [Dapr Workflow](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-overview/). With this SDK, you can define, schedule, and manage durable orchestrations using ordinary Python code.
+This repo contains a Python SDK for use with the [Azure Durable Task Scheduler](https://techcommunity.microsoft.com/blog/appsonazureblog/announcing-limited-early-access-of-the-durable-task-scheduler-for-azure-durable-/4286526) and the [Durable Task Framework for Go](https://github.com/microsoft/durabletask-go). With this SDK, you can define, schedule, and manage durable orchestrations using ordinary Python code.
 
 ⚠️ **This SDK is currently under active development and is not yet ready for production use.** ⚠️
 
-> Note that this project is **not** currently affiliated with the [Durable Functions](https://docs.microsoft.com/azure/azure-functions/durable/durable-functions-overview) project for Azure Functions. If you are looking for a Python SDK for Durable Functions, please see [this repo](https://github.com/Azure/azure-functions-durable-python).
-
+> Note that this SDK is **not** currently compatible with [Azure Durable Functions](https://docs.microsoft.com/azure/azure-functions/durable/durable-functions-overview). If you are looking for a Python SDK for Azure Durable Functions, please see [this repo](https://github.com/Azure/azure-functions-durable-python).
 
 ## Supported patterns
 

durabletask-azuremanaged/durabletask/azuremanaged/client.py

@@ -1,29 +1,30 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from typing import Optional
-from durabletask.client import TaskHubGrpcClient, OrchestrationStatus
-from durabletask.azuremanaged.internal.access_token_manager import AccessTokenManager
-from durabletask.azuremanaged.durabletask_grpc_interceptor import DTSDefaultClientInterceptorImpl
 from azure.core.credentials import TokenCredential
 
+from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import \
+    DTSDefaultClientInterceptorImpl
+from durabletask.client import TaskHubGrpcClient
+
+
 # Client class used for Durable Task Scheduler (DTS)
 class DurableTaskSchedulerClient(TaskHubGrpcClient):
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
                  token_credential: TokenCredential,
-                 secure_channel: Optional[bool] = True):
+                 secure_channel: bool = True):
 
-        if taskhub == None:
+        if not taskhub:
             raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
 
-        self._interceptors = [DTSDefaultClientInterceptorImpl(token_credential, taskhub)]
+        interceptors = [DTSDefaultClientInterceptorImpl(token_credential, taskhub)]
 
         # We pass in None for the metadata so we don't construct an additional interceptor in the parent class
         # Since the parent class doesn't use anything metadata for anything else, we can set it as None
         super().__init__(
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
-            interceptors=self._interceptors)
+            interceptors=interceptors)

durabletask-azuremanaged/durabletask/azuremanaged/internal/access_token_manager.py

@@ -1,24 +1,33 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
-from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
 from datetime import datetime, timedelta, timezone
 from typing import Optional
+
+from azure.core.credentials import AccessToken, TokenCredential
+
 import durabletask.internal.shared as shared
-from azure.core.credentials import TokenCredential
+
 
 # By default, when there's 10minutes left before the token expires, refresh the token
 class AccessTokenManager:
-    def __init__(self, refresh_interval_seconds: int = 600, token_credential: TokenCredential = None):
+
+    _token: Optional[AccessToken]
+
+    def __init__(self, token_credential: Optional[TokenCredential], refresh_interval_seconds: int = 600):
         self._scope = "https://durabletask.io/.default"
         self._refresh_interval_seconds = refresh_interval_seconds
         self._logger = shared.get_logger("token_manager")
 
         self._credential = token_credential
-        
-        self._token = self._credential.get_token(self._scope)
-        self.expiry_time = None
 
-    def get_access_token(self) -> str:
+        if self._credential is not None:
+            self._token = self._credential.get_token(self._scope)
+            self.expiry_time = datetime.fromtimestamp(self._token.expires_on, tz=timezone.utc)
+        else:
+            self._token = None
+            self.expiry_time = None
+
+    def get_access_token(self) -> Optional[AccessToken]:
         if self._token is None or self.is_token_expired():
             self.refresh_token()
         return self._token
@@ -32,9 +41,9 @@ def is_token_expired(self) -> bool:
         return datetime.now(timezone.utc) >= (self.expiry_time - timedelta(seconds=self._refresh_interval_seconds))
 
     def refresh_token(self):
-        new_token = self._credential.get_token(self._scope)
-        self._token = f"Bearer {new_token.token}"
-        
-        # Convert UNIX timestamp to timezone-aware datetime
-        self.expiry_time = datetime.fromtimestamp(new_token.expires_on, tz=timezone.utc)
-        self._logger.debug(f"Token refreshed. Expires at: {self.expiry_time}")
+        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._logger.debug(f"Token refreshed. Expires at: {self.expiry_time}")

durabletask-azuremanaged/durabletask/azuremanaged/durabletask_grpc_interceptor.py renamed to durabletask-azuremanaged/durabletask/azuremanaged/internal/durabletask_grpc_interceptor.py

@@ -1,35 +1,41 @@
-# Copyright (c) Microsoft Corporation.
-# Licensed under the MIT License.
-
-from durabletask.internal.grpc_interceptor import _ClientCallDetails, DefaultClientInterceptorImpl
-from durabletask.azuremanaged.internal.access_token_manager import AccessTokenManager
-from azure.core.credentials import TokenCredential
-import grpc
-
-class DTSDefaultClientInterceptorImpl (DefaultClientInterceptorImpl):
-    """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
-    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an 
-    interceptor to add additional headers to all calls as needed."""
-
-    def __init__(self, token_credential: TokenCredential, taskhub_name: str):
-        self._metadata = [("taskhub", taskhub_name)]
-        super().__init__(self._metadata)
-        
-        if token_credential is not None:
-            self._token_credential = token_credential
-            self._token_manager = AccessTokenManager(token_credential=self._token_credential)
-            token = self._token_manager.get_access_token()
-            self._metadata.append(("authorization", token))
-
-    def _intercept_call(
-                self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
-        """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
-            call details."""
-        # Refresh the auth token if it is present and needed
-        if self._metadata is not None:
-            for i, (key, _) in enumerate(self._metadata):
-                if key.lower() == "authorization":  # Ensure case-insensitive comparison
-                    new_token = self._token_manager.get_access_token()  # Get the new token
-                    self._metadata[i] = ("authorization", new_token)  # Update the token
-
-        return super()._intercept_call(client_call_details)
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import grpc
+from azure.core.credentials import TokenCredential
+
+from durabletask.azuremanaged.internal.access_token_manager import \
+    AccessTokenManager
+from durabletask.internal.grpc_interceptor import (
+    DefaultClientInterceptorImpl, _ClientCallDetails)
+
+
+class DTSDefaultClientInterceptorImpl (DefaultClientInterceptorImpl):
+    """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
+    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an
+    interceptor to add additional headers to all calls as needed."""
+
+    def __init__(self, token_credential: TokenCredential, taskhub_name: str):
+        self._metadata = [("taskhub", taskhub_name)]
+        super().__init__(self._metadata)
+
+        if token_credential is not None:
+            self._token_credential = token_credential
+            self._token_manager = AccessTokenManager(token_credential=self._token_credential)
+            access_token = self._token_manager.get_access_token()
+            if access_token is not None:
+                self._metadata.append(("authorization", f"Bearer {access_token.token}"))
+
+    def _intercept_call(
+            self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
+        """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
+            call details."""
+        # Refresh the auth token if it is present and needed
+        if self._metadata is not None:
+            for i, (key, _) in enumerate(self._metadata):
+                if key.lower() == "authorization":  # Ensure case-insensitive comparison
+                    new_token = self._token_manager.get_access_token()  # Get the new token
+                    if new_token is not None:
+                        self._metadata[i] = ("authorization", f"Bearer {new_token.token}")  # Update the token
+
+        return super()._intercept_call(client_call_details)

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -1,22 +1,23 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from typing import Optional
-from durabletask.worker import TaskHubGrpcWorker
-from durabletask.azuremanaged.internal.access_token_manager import AccessTokenManager
-from durabletask.azuremanaged.durabletask_grpc_interceptor import DTSDefaultClientInterceptorImpl
 from azure.core.credentials import TokenCredential
 
+from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import \
+    DTSDefaultClientInterceptorImpl
+from durabletask.worker import TaskHubGrpcWorker
+
+
 # Worker class used for Durable Task Scheduler (DTS)
 class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
                  token_credential: TokenCredential,
-                 secure_channel: Optional[bool] = True):
-        
-        if taskhub == None:
-            raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
+                 secure_channel: bool = True):
+
+        if not taskhub:
+            raise ValueError("The taskhub value cannot be empty.")
 
         interceptors = [DTSDefaultClientInterceptorImpl(token_credential, taskhub)]
 
@@ -25,5 +26,5 @@ def __init__(self, *,
         super().__init__(
             host_address=host_address,
             secure_channel=secure_channel,
-            metadata=None, 
-            interceptors=interceptors)
+            metadata=None,
+            interceptors=interceptors)

durabletask/client.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
-from typing import Any, Optional, TypeVar, Union
+from typing import Any, Optional, Sequence, TypeVar, Union
 
 import grpc
 from google.protobuf import wrappers_pb2
@@ -15,9 +15,8 @@
 import durabletask.internal.orchestrator_service_pb2 as pb
 import durabletask.internal.orchestrator_service_pb2_grpc as stubs
 import durabletask.internal.shared as shared
-from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
-
 from durabletask import task
+from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
 
 TInput = TypeVar('TInput')
 TOutput = TypeVar('TOutput')
@@ -99,22 +98,23 @@ def __init__(self, *,
                  log_handler: Optional[logging.Handler] = None,
                  log_formatter: Optional[logging.Formatter] = None,
                  secure_channel: bool = False,
-                 interceptors: Optional[list[Union[grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor]]] = None):
+                 interceptors: Optional[Sequence[shared.ClientInterceptor]] = None):
 
-        # Determine the interceptors to use
+        # If the caller provided metadata, we need to create a new interceptor for it and
+        # add it to the list of interceptors.
         if interceptors is not None:
-            self._interceptors = interceptors
-            if metadata:
-                self._interceptors.append(DefaultClientInterceptorImpl(metadata))
-        elif metadata:
-            self._interceptors = [DefaultClientInterceptorImpl(metadata)]
+            interceptors = list(interceptors)
+            if metadata is not None:
+                interceptors.append(DefaultClientInterceptorImpl(metadata))
+        elif metadata is not None:
+            interceptors = [DefaultClientInterceptorImpl(metadata)]
         else:
-            self._interceptors = None
+            interceptors = None
 
         channel = shared.get_grpc_channel(
             host_address=host_address,
             secure_channel=secure_channel,
-            interceptors=self._interceptors
+            interceptors=interceptors
         )
         self._stub = stubs.TaskHubSidecarServiceStub(channel)
         self._logger = shared.get_logger("client", log_handler, log_formatter)
@@ -134,7 +134,7 @@ def schedule_new_orchestration(self, orchestrator: Union[task.Orchestrator[TInpu
             scheduledStartTimestamp=helpers.new_timestamp(start_at) if start_at else None,
             version=wrappers_pb2.StringValue(value=""),
             orchestrationIdReusePolicy=reuse_id_policy,
-            )
+        )
 
         self._logger.info(f"Starting new '{name}' instance with ID = '{req.instanceId}'.")
         res: pb.CreateInstanceResponse = self._stub.StartInstance(req)

durabletask/internal/grpc_interceptor.py

@@ -19,28 +19,28 @@ class _ClientCallDetails(
 
 
 class DefaultClientInterceptorImpl (
-    grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
-    grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor):
+        grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
+        grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor):
     """The class implements a UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor,
-    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an 
+    StreamUnaryClientInterceptor and StreamStreamClientInterceptor from grpc to add an
     interceptor to add additional headers to all calls as needed."""
 
     def __init__(self, metadata: list[tuple[str, str]]):
         super().__init__()
         self._metadata = metadata
 
     def _intercept_call(
-                self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
+            self, client_call_details: _ClientCallDetails) -> grpc.ClientCallDetails:
         """Internal intercept_call implementation which adds metadata to grpc metadata in the RPC
             call details."""
         if self._metadata is None:
             return client_call_details
-        
+
         if client_call_details.metadata is not None:
             metadata = list(client_call_details.metadata)
         else:
             metadata = []
-        
+
         metadata.extend(self._metadata)
         client_call_details = _ClientCallDetails(
             client_call_details.method, client_call_details.timeout, metadata,

durabletask/internal/shared.py

@@ -5,11 +5,17 @@
 import json
 import logging
 from types import SimpleNamespace
-from typing import Any, Optional
-from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl
+from typing import Any, Optional, Sequence, Union
 
 import grpc
 
+ClientInterceptor = Union[
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor
+]
+
 # Field name used to indicate that an object was automatically serialized
 # and should be deserialized as a SimpleNamespace
 AUTO_SERIALIZED = "__durabletask_autoobject__"
@@ -25,8 +31,8 @@ def get_default_host_address() -> str:
 def get_grpc_channel(
         host_address: Optional[str],
         secure_channel: bool = False,
-        interceptors: Optional[list[DefaultClientInterceptorImpl]] = None) -> grpc.Channel:
-    
+        interceptors: Optional[Sequence[ClientInterceptor]] = None) -> grpc.Channel:
+
     if host_address is None:
         host_address = get_default_host_address()
 
@@ -55,6 +61,7 @@ def get_grpc_channel(
         channel = grpc.intercept_channel(channel, *interceptors)
     return channel
 
+
 def get_logger(
         name_suffix: str,
         log_handler: Optional[logging.Handler] = None,
@@ -99,7 +106,7 @@ def default(self, obj):
         if dataclasses.is_dataclass(obj):
             # Dataclasses are not serializable by default, so we convert them to a dict and mark them for
             # automatic deserialization by the receiver
-            d = dataclasses.asdict(obj)  # type: ignore 
+            d = dataclasses.asdict(obj)  # type: ignore
             d[AUTO_SERIALIZED] = True
             return d
         elif isinstance(obj, SimpleNamespace):