design-patterns/python

68 notes in this subject, read out of this brain and free to use. This is page 2 of 2.

MapReduce Tree Python implementation

Python implementation using LeafWorkflow and NodeWorkflow classes. LeafWorkflow executes process_leaf Activity and signals result using workflow.get_external_workflow_handle(parent_workflow_id).signal(). NodeWorkflow uses @workflow.signal decorator for node_result handler, checks len(records) against LEAF_THRESHOLD, starts child workflows using workflow.start_child_workflow(), and waits using workflow.wait_condition(lambda: self._received >= expected).

Pick First pattern implementation in Python

In Python, the Pick First pattern uses `asyncio.create_task()` to start Activities concurrently. `workflow.wait()` with `return_when=asyncio.FIRST_COMPLETED` waits for the first Activity to complete. After capturing the result, pending tasks are cancelled explicitly by calling `task.cancel()` on each. To wait for cancellation cleanup, set `cancellation_type=workflow.ActivityCancellationType.WAIT_CANCELLATION_COMPLETED` when calling `workflow.execute_activity()`, then await pending tasks while catching `asyncio.CancelledError`.

Retry Alerting via Metrics implementation in Python

```python from temporalio import activity from temporalio.exceptions import ApplicationError ALERT_THRESHOLD = 5 @activity.defn async def call_downstream_service(endpoint: str) -> str: info = activity.info() if info.attempt > ALERT_THRESHOLD: meter = activity.metric_meter() meter.create_counter( "high_activity_error_count", "Activity has exceeded the failure attempt threshold", ).add(1) response = await downstream.call(endpoint) return response.data ``` This example shows reading the attempt number from activity info and emitting a counter metric when it exceeds the threshold.

Retry Alerting with dimension tags in Python

```python if info.attempt > ALERT_THRESHOLD: meter = activity.metric_meter() meter.create_counter( "high_activity_error_count", "Activity has exceeded the failure attempt threshold", ).add(1, {"activity_type": info.activity_type, "endpoint": endpoint}) ``` Add tags to identify which Activity type, endpoint, or Workflow is producing high attempt counts.

Parallel Execution in Python

In Python, workflow.execute_activity() returns awaitables. Use asyncio.gather() to wait for all of them to complete.

Python parallel Activities example

import asyncio from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import process @workflow.defn class ParallelWorkflow: @workflow.run async def run(self, items: list[str]) -> list[str]: results = await asyncio.gather( *[ workflow.execute_activity( process, item, start_to_close_timeout=timedelta(seconds=30), ) for item in items ] ) return list(results) This example starts one Activity per item in a list and waits for all of them to complete.

Python batch processing example

import asyncio from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import process @workflow.defn class BatchWorkflow: @workflow.run async def run(self, items: list[str], max_parallel: int) -> list[str]: results: list[str] = [] for i in range(0, len(items), max_parallel): batch = items[i : i + max_parallel] batch_results = await asyncio.gather( *[ workflow.execute_activity( process, item, start_to_close_timeout=timedelta(seconds=30), ) for item in batch ] ) results.extend(batch_results) return results This example implements controlled parallelism by processing items in batches.

Error handling in parallel execution - Python

import asyncio from dataclasses import dataclass from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import process @dataclass class Result: item: str output: str | None = None error: str | None = None @workflow.defn class ResilientParallelWorkflow: @workflow.run async def run(self, items: list[str]) -> list[Result]: tasks = [ workflow.execute_activity( process, item, start_to_close_timeout=timedelta(seconds=30), ) for item in items ] outcomes = await asyncio.gather(*tasks, return_exceptions=True) results: list[Result] = [] for item, outcome in zip(items, outcomes): if isinstance(outcome, BaseException): results.append(Result(item=item, error=str(outcome))) else: results.append(Result(item=item, output=outcome)) return results This example wraps each Activity in error handling using asyncio.gather() with return_exceptions=True so that individual failures do not prevent other Activities from completing.

Give your agent this brain