design-patterns/python

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

Python Activity Dependency Injection implementation

In Python, Activities are defined as a dataclass with @activity.defn methods. Dependencies are passed as fields through the dataclass. At Worker startup, instantiate the dataclass with real implementations and register each Activity method separately with the Worker. Workflows use workflow.execute_activity_method to reference Activity methods, which resolves to the registered instance.

Pitfall: Registering class methods as static in Python

If you register BotService.send_message (the unbound method) instead of bot_service.send_message (a method on an instance), the self parameter is not bound, causing a missing argument error at runtime.

Python circuit breaker example with dependency injection

# activities.py — pybreaker import pybreaker from dataclasses import dataclass from temporalio import activity @dataclass class PaymentActivities: payment_api: PaymentAPI breaker: pybreaker.CircuitBreaker @activity.defn async def charge_customer(self, order_id: str, amount: int) -> str: activity.logger.info( "Charging customer", extra={"order_id": order_id, "amount": amount} ) # Raises pybreaker.CircuitBreakerError when the breaker is open. return await self.breaker.call_async( self.payment_api.charge, order_id, amount ) # worker.py import asyncio import pybreaker from temporalio.client import Client from temporalio.worker import Worker from activities import PaymentActivities, PaymentAPI from workflows import PaymentWorkflow async def main(): client = await Client.connect("localhost:7233") # Construct the breaker once at Worker startup so its failure # counters are shared across all Activity executions. breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30) payment_activities = PaymentActivities( payment_api=PaymentAPI("https://api.example.com"), breaker=breaker, ) worker = Worker( client, task_queue="payment", workflows=[PaymentWorkflow], activities=[ payment_activities.charge_customer, payment_activities.send_receipt, ], ) await worker.run() if __name__ == "__main__": asyncio.run(main())

Python Activity Dependency Injection code example

# activities.py from dataclasses import dataclass from temporalio import activity @dataclass class PaymentActivities: db_client: DBClient email_client: EmailClient @activity.defn async def charge_customer(self, order_id: str, amount: int) -> str: activity.logger.info( "Charging customer", extra={"order_id": order_id, "amount": amount} ) receipt_id = await self.db_client.process_payment(order_id, amount) return receipt_id @activity.defn async def send_receipt(self, email: str, receipt_id: str) -> None: await self.email_client.send(email, "Payment Receipt", receipt_id) # workflows.py from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import PaymentActivities @workflow.defn class PaymentWorkflow: @workflow.run async def run(self, order_id: str, amount: int, email: str) -> None: receipt_id = await workflow.execute_activity_method( PaymentActivities.charge_customer, args=[order_id, amount], start_to_close_timeout=timedelta(seconds=30), ) await workflow.execute_activity_method( PaymentActivities.send_receipt, args=[email, receipt_id], start_to_close_timeout=timedelta(seconds=30), ) # worker.py import asyncio from temporalio.client import Client from temporalio.worker import Worker from activities import PaymentActivities from workflows import PaymentWorkflow async def main(): client = await Client.connect("localhost:7233") # Inject real dependencies at Worker startup payment_activities = PaymentActivities( db_client=PostgresClient("postgres://localhost:5432/payments"), email_client=SMTPClient("smtp://mail.example.com"), ) worker = Worker( client, task_queue="payment", workflows=[PaymentWorkflow], activities=[ payment_activities.charge_customer, payment_activities.send_receipt, ], ) await worker.run() if __name__ == "__main__": asyncio.run(main())

Batch Iterator Python implementation

from temporalio import workflow from temporalio.workflow import continue_as_new from datetime import timedelta from activities import fetch_page, process_record from shared import PAGE_SIZE @workflow.defn class BatchIteratorWorkflow: @workflow.run async def run(self, offset: int = 0, total_processed: int = 0) -> int: page = await workflow.execute_activity( fetch_page, args=[offset, PAGE_SIZE], start_to_close_timeout=timedelta(seconds=10), ) for record in page: await workflow.execute_activity( process_record, record, start_to_close_timeout=timedelta(seconds=10), ) total_processed += 1 workflow.logger.info( f"Processed page at offset {offset} ({len(page)} records, running total: {total_processed})" ) if len(page) == PAGE_SIZE: continue_as_new(args=[offset + PAGE_SIZE, total_processed]) return total_processed

Parallel child workflows in Python

To start multiple child workflows in parallel in Python, use asyncio.gather() with a list of workflow.execute_child_workflow() calls. This starts all children concurrently and awaits all results.

Synchronous child workflow execution in Python

To execute a child workflow synchronously in Python, use workflow.execute_child_workflow(ChildWorkflow.run, input, id="child-id"). This awaits until the child completes and returns a result, blocking the parent.

Asynchronous child workflow execution in Python

To start a child workflow asynchronously in Python, use workflow.start_child_workflow(ChildWorkflow.run, input, id="child-id", parent_close_policy=ParentClosePolicy.ABANDON). This returns a handle once the child has started without waiting for completion, allowing the parent to continue.

Python Continue-As-New implementation

In Python, call `workflow.continue_as_new(args=[cursor, total_processed])` to trigger a Continue-As-New transition. The method immediately stops the current execution and starts a new one. Use `workflow.info().is_continue_as_new_suggested()` to check if history is approaching the limit.

Delayed Start Python API

In Python, use `client.start_workflow()` with the `start_delay` parameter set to a `timedelta` object: ```python handle = await client.start_workflow( DelayedStartWorkflow.run, id=WORKFLOW_ID, task_queue=TASK_QUEUE, start_delay=timedelta(seconds=30), ) ```

Python activity for downstream rate limiting

Example of a simple Python Activity for downstream rate limiting: ```python # activities.py from temporalio import activity @activity.defn async def call_api(input: str) -> str: return await downstream_api.call(input) ```

Python worker with rate limiting

Example of a Python Worker configured with downstream rate limiting: ```python # worker.py from temporalio.worker import Worker from activities import call_api async def run_worker(client): worker = Worker( client, task_queue="rate-limited-tq", activities=[call_api], max_task_queue_activities_per_second=5.0, ) await worker.run() ``` This worker is dedicated to rate-limited activities and requires a separate worker registered on the workflow task queue.

Python workflow routing to rate-limited queue

Example of a Python Workflow routing Activities to a rate-limited queue: ```python # workflows.py from datetime import timedelta from temporalio import workflow from activities import call_api @workflow.defn class MyWorkflow: @workflow.run async def run(self, input: str) -> str: return await workflow.execute_activity( call_api, input, task_queue="rate-limited-tq", start_to_close_timeout=timedelta(seconds=30), ) ``` The Workflow specifies an explicit task_queue override in the Activity options to route the throttled Activity to the dedicated queue.

Early Return Python implementation example

Example showing Early Return pattern in Python using Update-with-Start: ```python # workflow.py from dataclasses import dataclass from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import init_transaction, complete_transaction, cancel_transaction @dataclass class TransactionRequest: amount: float currency: str @dataclass class Transaction: id: str status: str @workflow.defn class TransactionWorkflow: def __init__(self) -> None: self.tx: Transaction | None = None self.init_done = False self.init_err: Exception | None = None @workflow.run async def run(self, tx_request: TransactionRequest) -> Transaction | None: # Phase 1: Fast synchronous initialization (local activity) try: self.tx = await workflow.execute_local_activity( init_transaction, tx_request, schedule_to_close_timeout=timedelta(seconds=5), ) except Exception as e: self.init_err = e finally: self.init_done = True # Signal update handler # Phase 2: Slow asynchronous completion if self.init_err is not None: await workflow.execute_activity( cancel_transaction, self.tx, start_to_close_timeout=timedelta(seconds=30), ) return None await workflow.execute_activity( complete_transaction, self.tx, start_to_close_timeout=timedelta(seconds=30), ) return self.tx @workflow.update async def return_init_result(self) -> Transaction: await workflow.wait_condition(lambda: self.init_done) if self.init_err is not None: raise self.init_err return self.tx # client.py from temporalio.client import ( Client, WithStartWorkflowOperation, WorkflowUpdateStage, ) from temporalio.common import WorkflowIDConflictPolicy client = await Client.connect("localhost:7233") start_op = WithStartWorkflowOperation( TransactionWorkflow.run, tx_request, id="transaction-123", task_queue="transactions", id_conflict_policy=WorkflowIDConflictPolicy.FAIL, ) update_handle = await client.start_update_with_start_workflow( TransactionWorkflow.return_init_result, wait_for_stage=WorkflowUpdateStage.COMPLETED, start_workflow_operation=start_op, ) # Get initialization result immediately tx = await update_handle.result() # Use transaction ID immediately while workflow continues print(f"Transaction initialized: {tx.id}") ```

Entity Workflow Python implementation

from dataclasses import dataclass from datetime import datetime, timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import validate_profile @dataclass class UserState: status: str = "ACTIVE" profile: ProfileData | None = None pending_email: str | None = None created_at: datetime | None = None updated_at: datetime | None = None @dataclass class UserAccountInput: user_id: str state: UserState | None = None @workflow.defn class UserAccountWorkflow: def __init__(self) -> None: self.state = UserState(created_at=datetime.utcnow()) self.deleted = False self.operation_count = 0 @workflow.run async def run(self, input: UserAccountInput) -> None: if input.state is not None: self.state = input.state await workflow.wait_condition( lambda: self.deleted or workflow.info().is_continue_as_new_suggested() ) if not self.deleted and workflow.info().is_continue_as_new_suggested(): await workflow.wait_condition(workflow.all_handlers_finished) workflow.continue_as_new( UserAccountInput(user_id=input.user_id, state=self.state) ) self.state.status = "DELETED" @workflow.update async def update_profile(self, data: ProfileData) -> None: if self.deleted: raise ValueError("User account is deleted") await workflow.execute_activity( validate_profile, data, start_to_close_timeout=timedelta(seconds=30), ) self.state.profile = data self.state.updated_at = datetime.utcnow() self.operation_count += 1 @workflow.update async def suspend(self) -> None: if not self.deleted and self.state.status != "SUSPENDED": self.state.status = "SUSPENDED" self.state.updated_at = datetime.utcnow() self.operation_count += 1 @workflow.update async def reactivate(self) -> None: if not self.deleted and self.state.status == "SUSPENDED": self.state.status = "ACTIVE" self.state.updated_at = datetime.utcnow() self.operation_count += 1 @workflow.signal def delete(self) -> None: self.deleted = True @workflow.query def get_state(self) -> UserState: return self.state

Fan-Out Python implementation

```python # workflows.py from datetime import timedelta from temporalio import workflow from temporalio.workflow import ChildWorkflowHandle import asyncio from activities import process_record from shared import TASK_QUEUE, CHUNK_SIZE @workflow.defn class RecordBatchWorkflow: @workflow.run async def run(self, offset: int, length: int) -> int: processed = 0 for i in range(offset, offset + length): await workflow.execute_activity( process_record, i, start_to_close_timeout=timedelta(seconds=10), ) processed += 1 return processed @workflow.defn class FanOutWorkflow: @workflow.run async def run(self, total_records: int, chunk_size: int = CHUNK_SIZE) -> int: handles: list[ChildWorkflowHandle] = [] parent_id = workflow.info().workflow_id offset = 0 while offset < total_records: length = min(chunk_size, total_records - offset) handle = await workflow.start_child_workflow( RecordBatchWorkflow.run, args=[offset, length], id=f"{parent_id}/batch-{offset}", task_queue=TASK_QUEUE, ) handles.append(handle) offset += chunk_size results = await asyncio.gather(*handles) return sum(results) ```

Fast/Slow Retries Python example

```python from datetime import timedelta from temporalio import workflow from temporalio.common import RetryPolicy from temporalio.exceptions import ActivityError import activities @workflow.defn class FastSlowRetryWorkflow: @workflow.run async def run(self, request: str) -> str: # Phase 1: fast retries fast_policy = RetryPolicy( initial_interval=timedelta(seconds=1), backoff_coefficient=1.5, maximum_interval=timedelta(seconds=30), maximum_attempts=10, ) try: return await workflow.execute_activity( activities.call_downstream, request, start_to_close_timeout=timedelta(seconds=30), retry_policy=fast_policy, ) except ActivityError: workflow.logger.warning( "Fast retries exhausted — switching to slow retry phase", extra={"request": request}, ) # Phase 2: slow retries slow_policy = RetryPolicy( initial_interval=timedelta(minutes=5), backoff_coefficient=1.0, ) return await workflow.execute_activity( activities.call_downstream, request, start_to_close_timeout=timedelta(seconds=30), retry_policy=slow_policy, ) ``` This example shows a Workflow implementing fast retries with 1-second initial interval and 10 max attempts, transitioning to slow retries with 5-minute interval and unlimited attempts when fast phase is exhausted.

Python Fixed Count Retries example

from datetime import timedelta from temporalio import workflow from temporalio.common import RetryPolicy from temporalio.exceptions import ActivityError, RetryState import activities @workflow.defn class PaymentWorkflow: @workflow.run async def run(self, order_id: str) -> str: try: return await workflow.execute_activity( activities.charge_payment_api, order_id, start_to_close_timeout=timedelta(seconds=10), retry_policy=RetryPolicy(maximum_attempts=3), ) except ActivityError as e: if e.retry_state == RetryState.MAXIMUM_ATTEMPTS_REACHED: workflow.logger.error( "Payment failed: all 3 attempts exhausted", extra={"order_id": order_id}, ) raise This example shows how to set maximum_attempts=3 on a RetryPolicy for a Python workflow and catch ActivityError when retries are exhausted.

Fixed Wall-Time Retries Python implementation

Example of enforcing a 2-minute SLA with per-attempt 30-second timeout: ```python @workflow.defn class PaymentAuthWorkflow: @workflow.run async def run(self, transaction_id: str) -> str: try: return await workflow.execute_activity( activities.authorize_transaction, transaction_id, schedule_to_close_timeout=timedelta(minutes=2), # total budget start_to_close_timeout=timedelta(seconds=30), # per attempt retry_policy=RetryPolicy( initial_interval=timedelta(seconds=5), backoff_coefficient=1.5, maximum_interval=timedelta(seconds=30), ), ) except ActivityError as e: cause = e.__cause__ if isinstance(cause, TimeoutError) and cause.type == TimeoutType.SCHEDULE_TO_CLOSE: workflow.logger.error( "Authorization failed — 2-minute SLA breached", extra={"transaction_id": transaction_id}, ) raise ```

Short SLA without per-attempt timeout Python

For a 30-second authorization window, omit StartToCloseTimeout and let ScheduleToCloseTimeout act as the only bound: ```python result = await workflow.execute_activity( activities.authorize_transaction, transaction_id, schedule_to_close_timeout=timedelta(seconds=30), retry_policy=RetryPolicy( initial_interval=timedelta(seconds=3), backoff_coefficient=1.5, ), ) ```

Local Activity Python API

In Python, use `workflow.execute_local_activity()` to run Activity functions in-process. The function takes the activity function, arguments, and `schedule_to_close_timeout` option.

Local Activity Python example

from temporalio import workflow from datetime import timedelta from activities import validate_transaction, reserve_funds, settle_transaction LOCAL_ACTIVITY_TIMEOUT = timedelta(seconds=10) @workflow.defn class TransactionWorkflow: @workflow.run async def run(self, req: TransactionRequest) -> Transaction: # All three activities run in-process — no server round-trips. tx = await workflow.execute_local_activity( validate_transaction, req, schedule_to_close_timeout=LOCAL_ACTIVITY_TIMEOUT, ) tx = await workflow.execute_local_activity( reserve_funds, tx, schedule_to_close_timeout=LOCAL_ACTIVITY_TIMEOUT, ) return await workflow.execute_local_activity( settle_transaction, tx, schedule_to_close_timeout=LOCAL_ACTIVITY_TIMEOUT, )

Python Activity Heartbeat - basic progress tracking

Example showing how to process a large file line by line in Python, heartbeating every 100 lines and resuming from the last checkpoint on retry using activity.info().heartbeat_details.

Python Activity Heartbeat - cancellation handling

In Python, cancellation is delivered as an asyncio.CancelledError. The Activity should catch this exception, perform cleanup via cleanupResources(), and re-raise the error. Heartbeat calls and async operations become sensitive to cancellation when the context is cancelled.

Python ApplicationError with non_retryable flag

In Python, use temporalio.exceptions.ApplicationError with the non_retryable=True flag to mark an error as non-retryable at the throw site. Pass a type parameter to identify the error type. Example: raise ApplicationError(f"Order {order_id} not found", type="OrderNotFoundError", non_retryable=True)

Python RetryPolicy with non_retryable_error_types

In Python, pass a RetryPolicy to workflow.execute_activity() with non_retryable_error_types set to a list of error type name strings. Example: retry_policy=RetryPolicy(non_retryable_error_types=["OrderNotFoundError", "ValidationError"])

Catching and handling ActivityError in Python workflows

In Python, catch temporalio.exceptions.ActivityError in a try-except block. Access the underlying cause via the __cause__ attribute. Check if the cause is an ApplicationError and inspect its type property to route to the appropriate compensation or escalation path.

Frequent polling example in Python

```python # activities.py from temporalio import activity import asyncio @activity.defn async def do_poll() -> str: while True: activity.heartbeat() result = await external_service.check_status() if result == "COMPLETED": return result await asyncio.sleep(1) # workflows.py from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import do_poll @workflow.defn class FrequentPollingWorkflow: @workflow.run async def run(self) -> str: return await workflow.execute_activity( do_poll, start_to_close_timeout=timedelta(seconds=60), heartbeat_timeout=timedelta(seconds=2), ) ``` This example shows a frequent polling Activity that loops indefinitely with heartbeats every iteration, and a Workflow that executes the Activity with a 60-second start-to-close timeout and 2-second heartbeat timeout.

Infrequent polling example in Python

```python # activities.py from temporalio import activity from temporalio.exceptions import ApplicationError @activity.defn async def do_poll() -> str: result = await external_service.check_status() if result != "COMPLETED": raise ApplicationError("Service not ready, will retry") return result # workflows.py from datetime import timedelta from temporalio import workflow from temporalio.common import RetryPolicy with workflow.unsafe.imports_passed_through(): from activities import do_poll @workflow.defn class InfrequentPollingWorkflow: @workflow.run async def run(self) -> str: return await workflow.execute_activity( do_poll, start_to_close_timeout=timedelta(seconds=2), retry_policy=RetryPolicy( backoff_coefficient=1, initial_interval=timedelta(seconds=60), ), ) ``` This example shows an infrequent polling Activity that performs a single poll and throws if the service is not ready, and a Workflow that executes the Activity with a fixed 60-second retry interval (backoff_coefficient=1).

Periodic sequence polling example in Python

```python # workflows.py from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import do_poll @workflow.defn class PollingChildWorkflow: @workflow.run async def run(self, polling_interval_seconds: int) -> str: max_attempts = 10 for _ in range(max_attempts): result = await workflow.execute_activity( do_poll, start_to_close_timeout=timedelta(seconds=10), ) if result == "COMPLETED": return result await workflow.sleep(polling_interval_seconds) # Continue-as-new to prevent unbounded history workflow.continue_as_new(polling_interval_seconds) @workflow.defn class PeriodicPollingWorkflow: @workflow.run async def run(self) -> str: return await workflow.execute_child_workflow( PollingChildWorkflow.run, 5, id="ChildWorkflowPoll", ) ``` This example shows a Child Workflow that polls up to 10 times with 5-second intervals between attempts, then calls Continue-As-New to start a fresh execution. The parent Workflow executes the Child Workflow with a specific workflow ID.

Priority Task Queues example in Python

```python from temporalio.common import Priority handle = await client.start_workflow( ChargeCustomer.run, id="charge-customer-wf", task_queue="my-task-queue", priority=Priority(priority_key=1), ) ``` This example shows how to set Workflow priority at start time using the Python SDK.

Set Activity priority in Python

```python from temporalio.common import Priority # inside the workflow result = await workflow.execute_activity( process_payment, start_to_close_timeout=timedelta(minutes=1), priority=Priority(priority_key=1), ) ``` This example shows how to override an Activity's priority from the parent Workflow using the Python SDK.

Set Child Workflow priority in Python

```python from temporalio.common import Priority # inside the parent workflow result = await workflow.execute_child_workflow( ProcessOrder.run, id="process-order-child", task_queue="my-task-queue", priority=Priority(priority_key=2), ) ``` This example shows how to set a Child Workflow's priority using the Python SDK.

Python task assignment workflow with Updates

```python # workflows.py import uuid from dataclasses import dataclass from temporalio import workflow MAX_TASKS = 10 @dataclass class AssignmentResult: assignment_id: str task_name: str total_tasks: int @workflow.defn class TaskWorkflow: def __init__(self) -> None: self.tasks: list[str] = [] @workflow.run async def run(self) -> None: await workflow.wait_condition(lambda: False) @workflow.update async def assign_task(self, task_name: str) -> AssignmentResult: assignment_id = str(uuid.uuid4()) self.tasks.append(task_name) return AssignmentResult( assignment_id=assignment_id, task_name=task_name, total_tasks=len(self.tasks), ) @assign_task.validator def validate_assign_task(self, task_name: str) -> None: if len(self.tasks) >= MAX_TASKS: raise ValueError("Task limit reached") @workflow.query def get_tasks(self) -> list[str]: return list(self.tasks) ```

Resumable Activity Python implementation

Python Resumable Activity Workflow implementation: ```python from dataclasses import dataclass from datetime import timedelta from temporalio import workflow from temporalio.common import RetryPolicy, SearchAttributeKey from temporalio.exceptions import ActivityError import activities TRANSFER_STATUS_KEY = SearchAttributeKey.for_keyword("TransferStatus") @dataclass class TransferInput: from_account: str to_account: str amount: float @workflow.defn class TransferWorkflow: def __init__(self) -> None: self._status = "PENDING" self._corrected_account: str | None = None self._approval: bool | None = None @workflow.run async def run(self, transfer: TransferInput) -> str: account = transfer.to_account correction_attempts = 0 while True: self._status = "TRANSFERRING" try: result = await workflow.execute_activity( activities.execute_transfer, TransferInput(transfer.from_account, account, transfer.amount), start_to_close_timeout=timedelta(seconds=30), retry_policy=RetryPolicy(maximum_attempts=3), ) break except ActivityError: correction_attempts += 1 if correction_attempts > 5: self._status = "FAILED" workflow.upsert_search_attributes([TRANSFER_STATUS_KEY.value_set(self._status)]) raise self._status = "AWAITING_CORRECTION" workflow.upsert_search_attributes([TRANSFER_STATUS_KEY.value_set(self._status)]) workflow.logger.warning( "Transfer failed — waiting for account correction", extra={"to_account": account}, ) await workflow.wait_condition( lambda: self._corrected_account is not None ) account = self._corrected_account self._corrected_account = None self._status = "AWAITING_APPROVAL" await workflow.wait_condition(lambda: self._approval is not None) if self._approval: self._status = "COMPLETED" return f"Transfer of {transfer.amount} to {account} completed" self._status = "REJECTED" return "Transfer rejected by client" @workflow.signal def retry_with_correction(self, corrected_account: str) -> None: self._corrected_account = corrected_account @workflow.signal def approve(self, approved: bool) -> None: self._approval = approved @workflow.query def get_status(self) -> str: return self._status ``` This example shows how to maintain state as named fields, use Signal handlers to set fields, and block with wait_condition until they are non-null.

Resumable Activity: Python activity with non-retryable error

Python Activity implementation that distinguishes permanent from transient failures: ```python from temporalio import activity from temporalio.exceptions import ApplicationError @activity.defn async def execute_transfer(transfer: TransferInput) -> str: # Non-retryable: bad account number requires a human correction, not a retry. if not await account_service.exists(transfer.to_account): raise ApplicationError( f"Account {transfer.to_account} not found", type="AccountNotFoundError", non_retryable=True, ) # Other exceptions propagate as retryable so the RetryPolicy handles them. return await payment_service.transfer( transfer.from_account, transfer.to_account, transfer.amount ) ```

Signal with Start Python implementation

In Python, use client.start_workflow() with parameters: ShoppingCartWorkflow.run as the workflow function, id parameter for the workflow ID, task_queue parameter, start_signal parameter for the signal name, and start_signal_args parameter as a list containing the signal arguments. The example shows starting a shopping cart workflow with an AddItemSignal containing item_id, product_id, and quantity fields.

Python Saga pattern implementation

from temporalio import workflow @workflow.defn class OpenAccountWorkflow: @workflow.run async def run(self, req: OpenAccountRequest) -> str: compensations = [] try: await workflow.execute_activity( create_account, req, start_to_close_timeout=timedelta(seconds=10), ) compensations.append( lambda: workflow.execute_activity( clear_postal_addresses, req, start_to_close_timeout=timedelta(seconds=10), ) ) await workflow.execute_activity( add_address, req, start_to_close_timeout=timedelta(seconds=10), ) compensations.append( lambda: workflow.execute_activity( remove_client, req, start_to_close_timeout=timedelta(seconds=10), ) ) await workflow.execute_activity( add_client, req, start_to_close_timeout=timedelta(seconds=10), ) compensations.append( lambda: workflow.execute_activity( disconnect_bank_accounts, req, start_to_close_timeout=timedelta(seconds=10), ) ) await workflow.execute_activity( add_bank_account, req, start_to_close_timeout=timedelta(seconds=10), ) except Exception: for compensation in reversed(compensations): await compensation() raise This example shows how to implement the Saga pattern in Python using a compensations list and reversed() iteration on error.

Sliding Window Python implementation example

Example Python implementation of Sliding Window pattern: ```python from datetime import timedelta from temporalio import workflow from temporalio.exceptions import ApplicationError from temporalio.workflow import ParentClosePolicy, continue_as_new from activities import process_record from shared import COMPLETION_SIGNAL, TASK_QUEUE, WINDOW_SIZE, SlidingWindowInput @workflow.defn class RecordProcessorWorkflow: """Child Workflow: processes one record and signals the parent on completion.""" @workflow.run async def run(self, record_id: str) -> None: await workflow.execute_activity( process_record, record_id, start_to_close_timeout=timedelta(seconds=30), ) parent = workflow.get_external_workflow_handle(workflow.info().parent.workflow_id) try: await parent.signal(COMPLETION_SIGNAL, record_id) except ApplicationError as e: if "not found" not in str(e).lower(): raise @workflow.defn class SlidingWindowWorkflow: """Parent Workflow: maintains a fixed window of concurrent Child Workflows.""" def __init__(self) -> None: self._active = 0 self._total_processed = 0 @workflow.signal(name=COMPLETION_SIGNAL) def record_completed(self, record_id: str) -> None: self._active -= 1 self._total_processed += 1 @workflow.run async def run(self, input: SlidingWindowInput) -> int: self._total_processed += input.total_processed self._active += input.active record_ids = input.record_ids window_size = input.window_size parent_id = workflow.info().workflow_id next_index = input.start_index dispatched = 0 while next_index < len(record_ids): await workflow.wait_condition(lambda: self._active < window_size) await workflow.start_child_workflow( RecordProcessorWorkflow.run, record_ids[next_index], id=f"{parent_id}/record-{record_ids[next_index]}", task_queue=TASK_QUEUE, parent_close_policy=ParentClosePolicy.ABANDON, ) next_index += 1 dispatched += 1 self._active += 1 if dispatched >= window_size: continue_as_new(args=[SlidingWindowInput( record_ids=record_ids, window_size=window_size, start_index=next_index, total_processed=self._total_processed, active=self._active, )]) await workflow.wait_condition(lambda: self._active == 0) return self._total_processed ```

Updatable Timer Python implementation

```python # updatable_timer.py import asyncio from datetime import timedelta from temporalio import workflow class UpdatableTimer: def __init__(self, wake_up_time: float) -> None: self._wake_up_time = wake_up_time self._wake_up_time_updated = False async def sleep_until(self, wake_up_time: float) -> None: self._wake_up_time = wake_up_time while True: self._wake_up_time_updated = False sleep_secs = self._wake_up_time - workflow.time() try: await workflow.wait_condition( lambda: self._wake_up_time_updated, timeout=timedelta(seconds=max(sleep_secs, 0)), ) # Condition met: wake-up time was updated, loop to recalculate except asyncio.TimeoutError: break # Timer expired def update_wake_up_time(self, wake_up_time: float) -> None: self._wake_up_time = wake_up_time self._wake_up_time_updated = True # Unblocks wait_condition @property def wake_up_time(self) -> float: return self._wake_up_time ```

Updatable Timer basic approval workflow Python

```python # workflows.py import asyncio from datetime import timedelta from temporalio import workflow @workflow.defn class ApprovalWorkflow: def __init__(self) -> None: self._approved = False self._status = "PENDING" @workflow.run async def run(self, approval_deadline: float) -> None: timeout_secs = approval_deadline - workflow.time() try: await workflow.wait_condition( lambda: self._approved, timeout=timedelta(seconds=max(timeout_secs, 0)), ) self._status = "APPROVED" except asyncio.TimeoutError: self._status = "REJECTED" @workflow.signal def approve(self) -> None: self._approved = True @workflow.query def get_status(self) -> str: return self._status ``` This Workflow waits with both a deadline duration and a condition that checks the approved flag. If the approve Signal arrives before the deadline, the condition becomes true and the Workflow sets the status to APPROVED. If the deadline expires first, the Workflow sets the status to REJECTED.

Updatable Timer multi-extension approval workflow Python

```python # workflows.py from temporalio import workflow from .updatable_timer import UpdatableTimer @workflow.defn class MultiExtensionApprovalWorkflow: def __init__(self) -> None: self._timer = UpdatableTimer(0) self._approved = False self._rejected = False @workflow.run async def run(self, initial_deadline: float) -> None: await self._timer.sleep_until(initial_deadline) if not self._approved: self._rejected = True @workflow.signal def extend_deadline(self, new_deadline: float) -> None: if not self._approved and not self._rejected: self._timer.update_wake_up_time(new_deadline) @workflow.signal def approve(self) -> None: self._approved = True ``` The extendDeadline Signal handler checks that the Workflow has not already been approved or rejected before updating the timer. Each update unblocks the timer loop, which recalculates the remaining duration and blocks again.

Python Worker-Specific Task Queues example workflow

```python # workflows.py from datetime import timedelta from temporalio import workflow with workflow.unsafe.imports_passed_through(): from activities import download, process, upload @workflow.defn class FileProcessingWorkflow: @workflow.run async def run(self, source: str, destination: str) -> None: downloaded = await workflow.execute_activity( download, source, start_to_close_timeout=timedelta(seconds=20), ) processed = await workflow.execute_activity( process, downloaded.file_name, task_queue=downloaded.host_task_queue, schedule_to_start_timeout=timedelta(seconds=10), start_to_close_timeout=timedelta(seconds=20), ) await workflow.execute_activity( upload, args=[processed, destination], task_queue=downloaded.host_task_queue, schedule_to_start_timeout=timedelta(seconds=10), start_to_close_timeout=timedelta(seconds=20), ) ``` This example shows a workflow that downloads a file on any worker, then processes and uploads it on the same host-specific worker.

Python Worker-Specific Task Queues example activity

```python # activities.py from dataclasses import dataclass from temporalio import activity @dataclass class TaskQueueFileNamePair: host_task_queue: str file_name: str host_specific_task_queue: str = "" @activity.defn async def download(source: str) -> TaskQueueFileNamePair: local_file = await download_to_local_disk(source) return TaskQueueFileNamePair( host_task_queue=host_specific_task_queue, file_name=local_file, ) @activity.defn async def process(file_name: str) -> str: processed = await process_local_file(file_name) return processed @activity.defn async def upload(file_name: str, destination: str) -> None: await upload_from_local_disk(file_name, destination) ``` The download Activity returns both the file path and the host-specific Task Queue name. The process and upload methods operate on local files, which are guaranteed to exist because they run on the same host.

Python Worker-Specific Task Queues example worker setup

```python # worker.py import asyncio import uuid import socket from temporalio.client import Client from temporalio.worker import Worker from workflows import FileProcessingWorkflow from activities import download, process, upload import activities as act_module async def main(): client = await Client.connect("localhost:7233") default_task_queue = "FileProcessing" host_task_queue = f"FileProcessing-{socket.gethostname()}-{uuid.uuid4()}" act_module.host_specific_task_queue = host_task_queue default_worker = Worker( client, task_queue=default_task_queue, workflows=[FileProcessingWorkflow], activities=[download, process, upload], ) host_worker = Worker( client, task_queue=host_task_queue, activities=[download, process, upload], ) await asyncio.gather(default_worker.run(), host_worker.run()) if __name__ == "__main__": asyncio.run(main()) ``` Each Worker registers with both the default Task Queue and its own host-specific Task Queue. The host-specific queue name includes hostname and UUID for uniqueness.

Task Queue constant definition - Python example

In Python, define a Task Queue name constant in a shared module file (e.g., shared.py): TASK_QUEUE_NAME = "my-task-queue-name". Import and reference this constant in both the workflow client code and worker configuration code.

Patching code example - correct arrangement

if patched('v3'): # This is the newest version of the code. # put this at the top, so when it is running # a fresh execution and not replaying, # this patched statement will return true # and it will run the new code. pass elif patched('v2'): pass else: pass

Patching code example - incorrect arrangement

if patched('v2'): # This is bad because when doing a new execution (i.e. not replaying), # patched statements evaluate to True (and put a marker # in the event history), which means that new executions # will use v2, and miss v3 below pass elif patched('v3'): pass else: pass

Workflow Definition syntax in Python

A Workflow Definition in Python uses the @workflow.defn decorator on a class and @workflow.run decorator on the async method. Example: @workflow.defn class YourWorkflow: @workflow.run async def YourBasicWorkflow(self, input: str) -> str: ...

Eager Workflow Start Python example

import asyncio from temporalio.client import Client from temporalio.worker import Worker from workflows import TransactionWorkflow from activities import validate_transaction, settle_transaction from shared import TASK_QUEUE, TransactionRequest async def main(): client = await Client.connect("localhost:7233") async with Worker( client, task_queue=TASK_QUEUE, workflows=[TransactionWorkflow], activities=[validate_transaction, settle_transaction], ): result = await client.execute_workflow( TransactionWorkflow.run, TransactionRequest(amount=100.00, currency="USD"), id="eager-workflow-start-demo", task_queue=TASK_QUEUE, request_eager_start=True, ) print(f"Transaction complete: ID={result.id} Status={result.status}") if __name__ == "__main__": asyncio.run(main()) This example shows starting a Worker in the same process with the same client, then executing a Workflow with request_eager_start=True to dispatch the first WorkflowTask inline.

Early Return + Local Activities Python example

```python # workflows.py from temporalio import workflow from datetime import timedelta from activities import validate_transaction, init_transaction, complete_transaction, cancel_transaction LOCAL_TIMEOUT = timedelta(seconds=5) ACTIVITY_TIMEOUT = timedelta(seconds=30) @workflow.defn class TransactionWorkflow: def __init__(self) -> None: self._tx: Transaction | None = None self._phase1_done = False self._phase1_error: Exception | None = None @workflow.update async def get_result(self, req: TransactionRequest) -> Transaction: # Wait for Phase 1 to finish before returning to the caller. await workflow.wait_condition(lambda: self._phase1_done) if self._phase1_error: raise self._phase1_error return self._tx @workflow.run async def run(self, req: TransactionRequest) -> None: try: # Phase 1: Local Activities — zero server round-trips on the hot path. tx = await workflow.execute_local_activity( validate_transaction, req, schedule_to_close_timeout=LOCAL_TIMEOUT, ) self._tx = await workflow.execute_local_activity( init_transaction, tx, schedule_to_close_timeout=LOCAL_TIMEOUT, ) except Exception as e: self._phase1_error = e finally: self._phase1_done = True if self._phase1_error: if self._tx is not None: await workflow.execute_activity( cancel_transaction, self._tx, start_to_close_timeout=ACTIVITY_TIMEOUT, ) return # Phase 2: Regular Activities — background settlement (client already has response). await workflow.execute_activity( complete_transaction, self._tx, start_to_close_timeout=ACTIVITY_TIMEOUT, ) ```

Python Pattern 1 inbound webhook example

```python @workflow.defn class OrderWorkflow: def __init__(self) -> None: self._payment: Optional[PaymentPayload] = None @workflow.run async def run(self, order: OrderInput) -> str: workflow.logger.info(f"Order {order.order_id}: waiting for payment webhook") # Block until the inbound webhook signal arrives (or timeout after 24 hours) await workflow.wait_condition( lambda: self._payment is not None, timeout=timedelta(hours=24), ) if self._payment is None: return f"Order {order.order_id}: timed out waiting for payment" result = await workflow.execute_activity( process_payment, self._payment, start_to_close_timeout=timedelta(seconds=30), ) return result @workflow.signal async def payment_received(self, payload: PaymentPayload) -> None: workflow.logger.info(f"Payment signal received: {payload.payment_id}") self._payment = payload ``` This example shows an OrderWorkflow that uses wait_condition to block until a payment_received signal arrives, with a 24-hour timeout.

Python Pattern 1 Signal-with-Start starter example

```python # starter.py import asyncio import time from temporalio.client import Client from shared import TASK_QUEUE, OrderInput, PaymentPayload from workflows import OrderWorkflow async def main() -> None: client = await Client.connect("localhost:7233") order_id = f"order-{int(time.time() * 1000)}" order = OrderInput(order_id=order_id, amount=99.99) payment = PaymentPayload(payment_id=f"pay-{int(time.time() * 1000)}", amount=99.99) print(f"Sending webhook for order {order_id}") # Signal-with-Start: atomically starts the workflow (if not running) and # delivers the payment signal — this is exactly what your HTTP handler would do. handle = await client.start_workflow( OrderWorkflow.run, order, id=f"order-{order_id}", task_queue=TASK_QUEUE, start_signal="payment_received", start_signal_args=[payment], ) print(f"Webhook signal sent: {payment.payment_id}") result = await handle.result() print(f"Order completed: {result}") if __name__ == "__main__": asyncio.run(main()) ``` This example demonstrates Signal-with-Start which atomically creates the workflow if needed and delivers the payment signal in one call.

Python Pattern 2 delayed outbound callback example

```python @workflow.defn class DelayedCallbackWorkflow: @workflow.run async def run(self, input: CallbackInput) -> str: workflow.logger.info( f"Sleeping {input.delay_seconds}s before calling {input.callback_url}" ) # Durable sleep — survives worker restarts, server restarts, everything await workflow.sleep(timedelta(seconds=input.delay_seconds)) # Fire the outbound callback; Temporal retries on HTTP failure result = await workflow.execute_activity( send_webhook_callback, input, start_to_close_timeout=timedelta(minutes=5), ) workflow.logger.info(f"Callback delivered to {input.callback_url}") return result ``` This example shows a DelayedCallbackWorkflow using workflow.sleep() for a durable delay that survives restarts, followed by an activity to send the callback.

Python Pattern 3 async activity completion example

```python @activity.defn async def submit_job(input: JobInput) -> str: """Submit job to external system and return immediately. The activity completes asynchronously when the callback arrives.""" # Get the task token — this is the claim ticket task_token = activity.info().task_token # Submit the job to the external system, persisting the task token # so your callback handler can retrieve it later job_id = await external_service.submit( payload=input.payload, callback_url=f"https://your-api.example.com/callback", task_token_hex=task_token.hex(), # store alongside job_id ) activity.logger.info(f"Job {job_id} submitted; waiting for async callback") # Tell Temporal not to mark the activity complete on return; the external # callback will complete it later using the task token. activity.raise_complete_async() # In your webhook callback handler (e.g., FastAPI route): async def handle_callback(result: str, task_token_hex: str) -> None: token = bytes.fromhex(task_token_hex) client = await Client.connect("localhost:7233") handle = client.get_async_activity_handle(task_token=token) await handle.complete(result) # Workflow resumes with `result` immediately ``` This example shows Pattern 3 where an activity calls raise_complete_async() to signal it will complete via external callback, then the callback handler uses the task token to complete the activity.

Set fairness key and weight at Workflow start in Python

Set FairnessKey and FairnessWeight in the Priority object passed to client.start_workflow() with task_queue and priority parameters. Example: ```python from temporalio.common import Priority handle = await client.start_workflow( ProcessOrder.run, id="process-order-wf", task_queue="my-task-queue", priority=Priority( fairness_key="tenant-a", fairness_weight=2.0, ), ) ```

Set fairness key and weight on Activities in Python

Set FairnessKey and FairnessWeight in the Priority object passed to workflow.execute_activity(). Example: ```python from temporalio.common import Priority # inside the workflow result = await workflow.execute_activity( process_for_tenant, tenant_request, start_to_close_timeout=timedelta(minutes=1), priority=Priority( fairness_key="tenant-a", fairness_weight=2.0, ), ) ```

Use priority and fairness together in Python

Set both priority_key and fairness_key in the Priority object. Example: ```python from temporalio.common import Priority handle = await client.start_workflow( ChargeCustomer.run, id="charge-customer-wf", task_queue="my-task-queue", priority=Priority( priority_key=1, fairness_key="tenant-a", fairness_weight=2.0, ), ) ```

Event Accumulator Python implementation

Python implementation uses @workflow.signal(name=...) decorators and workflow.wait_condition() with a timedelta timeout. workflow.continue_as_new() accepts the new run arguments.

Event Accumulator Python example workflow

@workflow.defn class AccumulatorWorkflow: def __init__(self) -> None: self._unprocessed: deque[OrderItem] = deque() self._flush_requested = False @workflow.signal(name="add-item") async def add_item(self, item: OrderItem) -> None: self._unprocessed.append(item) @workflow.signal(name="flush") async def flush(self) -> None: self._flush_requested = True @workflow.run async def run( self, bucket_key: str, accumulated: list[OrderItem] | None = None, seen_keys: list[str] | None = None, ) -> str: items = list(accumulated or []) seen_set = set(seen_keys or []) while True: # Sliding window: wait for a signal or let the inactivity timer fire timed_out = not await workflow.wait_condition( lambda: bool(self._unprocessed) or self._flush_requested, timeout=timedelta(seconds=10), ) # Drain and deduplicate the signal queue while self._unprocessed: item = self._unprocessed.popleft() if item.order_id == bucket_key and item.item_id not in seen_set: seen_set.add(item.item_id) items.append(item) if timed_out or self._flush_requested: result = await workflow.execute_activity( process_items, args=[bucket_key, items], start_to_close_timeout=timedelta(seconds=10), ) if not self._unprocessed: return result # More signals arrived after timeout/flush — loop to process them if not self._unprocessed and workflow.info().is_continue_as_new_suggested(): workflow.continue_as_new(args=[bucket_key, items, sorted(seen_set)])

Give your agent this brain