final touches

Author: berndverst

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -5,19 +5,59 @@
 
 from azure.core.credentials import TokenCredential
 
-from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import (
-    DTSDefaultClientInterceptorImpl,
-)
-from durabletask.worker import TaskHubGrpcWorker
+from durabletask.azuremanaged.internal.durabletask_grpc_interceptor import \
+    DTSDefaultClientInterceptorImpl
+from durabletask.worker import ConcurrencyOptions, TaskHubGrpcWorker
 
 
 # Worker class used for Durable Task Scheduler (DTS)
 class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
+    """A worker implementation for Azure Durable Task Scheduler (DTS).
+
+    This class extends TaskHubGrpcWorker to provide integration with Azure's
+    Durable Task Scheduler service. It handles authentication via Azure credentials
+    and configures the necessary gRPC interceptors for DTS communication.
+
+    Args:
+        host_address (str): The gRPC endpoint address of the DTS service.
+        taskhub (str): The name of the task hub. Cannot be empty.
+        token_credential (Optional[TokenCredential]): Azure credential for authentication.
+            If None, anonymous authentication will be used.
+        secure_channel (bool, optional): Whether to use a secure gRPC channel (TLS).
+            Defaults to True.
+        concurrency_options (Optional[ConcurrencyOptions], optional): Configuration
+            for controlling worker concurrency limits. If None, default concurrency
+            settings will be used.
+
+    Raises:
+        ValueError: If taskhub is empty or None.
+
+    Example:
+        >>> from azure.identity import DefaultAzureCredential
+        >>> from durabletask.azuremanaged import DurableTaskSchedulerWorker
+        >>> from durabletask import ConcurrencyOptions
+        >>> 
+        >>> credential = DefaultAzureCredential()
+        >>> concurrency = ConcurrencyOptions(max_concurrent_activities=10)
+        >>> worker = DurableTaskSchedulerWorker(
+        ...     host_address="my-dts-service.azure.com:443",
+        ...     taskhub="my-task-hub",
+        ...     token_credential=credential,
+        ...     concurrency_options=concurrency
+        ... )
+
+    Note:
+        This worker automatically configures DTS-specific gRPC interceptors
+        for authentication and task hub routing. The parent class metadata
+        parameter is set to None since authentication is handled by the
+        DTS interceptor.
+    """
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
                  token_credential: Optional[TokenCredential],
-                 secure_channel: bool = True):
+                 secure_channel: bool = True,
+                 concurrency_options: Optional[ConcurrencyOptions] = None):
 
         if not taskhub:
             raise ValueError("The taskhub value cannot be empty.")
@@ -30,4 +70,5 @@ def __init__(self, *,
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
-            interceptors=interceptors)
+            interceptors=interceptors,
+            concurrency_options=concurrency_options)

durabletask-azuremanaged/pyproject.toml

@@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durabletask.azuremanaged"
-version = "0.1.5"
+version = "0.2.0"
 description = "Durable Task Python SDK provider implementation for the Azure Durable Task Scheduler"
 keywords = [
     "durable",

durabletask/worker.py

@@ -28,11 +28,10 @@
 
 
 class ConcurrencyOptions:
-    """Configuration options for controlling concurrency of different work item types.
+    """Configuration options for controlling concurrency of different work item types and the thread pool size.
 
-    This class mirrors the .NET DurableTask SDK's ConcurrencyOptions class,
-    providing fine-grained control over concurrent processing limits for
-    activities, orchestrations, and entities.
+    This class provides fine-grained control over concurrent processing limits for
+    activities, orchestrations and the thread pool size.
     """
 
     def __init__(
@@ -134,6 +133,83 @@ class ActivityNotRegisteredError(ValueError):
 
 
 class TaskHubGrpcWorker:
+    """A gRPC-based worker for processing durable task orchestrations and activities.
+
+    This worker connects to a Durable Task backend service via gRPC to receive and process
+    work items including orchestration functions and activity functions. It provides
+    concurrent execution capabilities with configurable limits and automatic retry handling.
+
+    The worker manages the complete lifecycle:
+    - Registers orchestrator and activity functions
+    - Connects to the gRPC backend service
+    - Receives work items and executes them concurrently
+    - Handles failures, retries, and state management
+    - Provides logging and monitoring capabilities
+
+    Args:
+        host_address (Optional[str], optional): The gRPC endpoint address of the backend service.
+            Defaults to the value from environment variables or localhost.
+        metadata (Optional[list[tuple[str, str]]], optional): gRPC metadata to include with
+            requests. Used for authentication and routing. Defaults to None.
+        log_handler (optional): Custom logging handler for worker logs. Defaults to None.
+        log_formatter (Optional[logging.Formatter], optional): Custom log formatter.
+            Defaults to None.
+        secure_channel (bool, optional): Whether to use a secure gRPC channel (TLS).
+            Defaults to False.
+        interceptors (Optional[Sequence[shared.ClientInterceptor]], optional): Custom gRPC
+            interceptors to apply to the channel. Defaults to None.
+        concurrency_options (Optional[ConcurrencyOptions], optional): Configuration for
+            controlling worker concurrency limits. If None, default settings are used.
+
+    Attributes:
+        concurrency_options (ConcurrencyOptions): The current concurrency configuration.
+
+    Example:
+        Basic worker setup:
+
+        >>> from durabletask import TaskHubGrpcWorker, ConcurrencyOptions
+        >>>
+        >>> # Create worker with custom concurrency settings
+        >>> concurrency = ConcurrencyOptions(
+        ...     maximum_concurrent_activity_work_items=50,
+        ...     maximum_concurrent_orchestration_work_items=20
+        ... )
+        >>> worker = TaskHubGrpcWorker(
+        ...     host_address="localhost:4001",
+        ...     concurrency_options=concurrency
+        ... )
+        >>>
+        >>> # Register functions
+        >>> @worker.add_orchestrator
+        ... def my_orchestrator(context, input):
+        ...     result = yield context.call_activity("my_activity", input="hello")
+        ...     return result
+        >>>
+        >>> @worker.add_activity 
+        ... def my_activity(context, input):
+        ...     return f"Processed: {input}"
+        >>>
+        >>> # Start the worker
+        >>> worker.start()
+        >>> # ... worker runs in background thread
+        >>> worker.stop()
+
+        Using as context manager:
+
+        >>> with TaskHubGrpcWorker() as worker:
+        ...     worker.add_orchestrator(my_orchestrator)
+        ...     worker.add_activity(my_activity)
+        ...     worker.start()
+        ...     # Worker automatically stops when exiting context
+
+    Raises:
+        RuntimeError: If attempting to add orchestrators/activities while the worker is running,
+            or if starting a worker that is already running.
+        OrchestratorNotRegisteredError: If an orchestration work item references an
+            unregistered orchestrator function.
+        ActivityNotRegisteredError: If an activity work item references an unregistered
+            activity function.
+    """
     _response_stream: Optional[grpc.Future] = None
     _interceptors: Optional[list[shared.ClientInterceptor]] = None