Temporal Developer

@workflow.defn
class OrderWorkflow:
    def __init__(self):
        self._approved = False
        self._items = []

    @workflow.signal
    async def approve(self) -> None:
        self._approved = True

    @workflow.signal
    async def add_item(self, item: str) -> None:
        self._items.append(item)

    @workflow.run
    async def run(self) -> str:
        # Wait for approval
        await workflow.wait_condition(lambda: self._approved)
        return f"Processed {len(self._items)} items"

@workflow.defn
class DynamicSignalWorkflow:
    def __init__(self):
        self._signals: dict[str, list[Any]] = {}

    @workflow.signal(dynamic=True)
    async def handle_signal(self, name: str, args: Sequence[RawValue]) -> None:
        if name not in self._signals:
            self._signals[name] = []
        self._signals[name].append(workflow.payload_converter().from_payload(args[0]))

@workflow.defn
class StatusWorkflow:
    def __init__(self):
        self._status = "pending"
        self._progress = 0

    @workflow.query
    def get_status(self) -> str:
        return self._status

    @workflow.query
    def get_progress(self) -> int:
        return self._progress

    @workflow.run
    async def run(self) -> str:
        self._status = "running"
        for i in range(100):
            self._progress = i
            await workflow.execute_activity(
                process_item, i,
                start_to_close_timeout=timedelta(minutes=1)
            )
        self._status = "completed"
        return "done"

@workflow.query(dynamic=True)
def handle_query(self, name: str, args: Sequence[RawValue]) -> Any:
    if name == "get_field":
        field_name = workflow.payload_converter().from_payload(args[0])
        return getattr(self, f"_{field_name}", None)

@workflow.defn
class OrderWorkflow:
    def __init__(self):
        self._items: list[str] = []

    @workflow.update
    async def add_item(self, item: str) -> int:
        self._items.append(item)
        return len(self._items)  # Returns new count to caller

    @add_item.validator
    def validate_add_item(self, item: str) -> None:
        if not item:
            raise ValueError("Item cannot be empty")
        if len(self._items) >= 100:
            raise ValueError("Order is full")

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, orders: list[Order]) -> list[str]:
        results = []
        for order in orders:
            result = await workflow.execute_child_workflow(
                ProcessOrderWorkflow.run,
                order,
                id=f"order-{order.id}",
                # Control what happens to child when parent completes
                parent_close_policy=workflow.ParentClosePolicy.ABANDON,
            )
            results.append(result)
        return results

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, target_workflow_id: str) -> None:
        # Get handle to external workflow
        handle = workflow.get_external_workflow_handle(target_workflow_id)

        # Signal the external workflow
        await handle.signal(TargetWorkflow.data_ready, data_payload)

        # Or cancel it
        await handle.cancel()

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, items: list[str]) -> list[str]:
        # Execute activities in parallel
        tasks = [
            workflow.execute_activity(
                process_item, item,
                start_to_close_timeout=timedelta(minutes=5)
            )
            for item in items
        ]
        return await asyncio.gather(*tasks)

# workflow.wait() - like asyncio.wait()
done, pending = await workflow.wait(
    futures,
    return_when=workflow.WaitConditionResult.FIRST_COMPLETED
)

# workflow.as_completed() - like asyncio.as_completed()
async for future in workflow.as_completed(futures):
    result = await future
    # Process each result as it completes

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, state: WorkflowState) -> str:
        while True:
            state = await process_batch(state)

            if state.is_complete:
                return "done"

            # Continue with fresh history before hitting limits
            if workflow.info().is_continue_as_new_suggested():
                workflow.continue_as_new(args=[state])

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self, order: Order) -> str:
        compensations: list[Callable[[], Awaitable[None]]] = []

        try:
            # Note - we save the compensation before running the activity,
            # because the following could happen:
            # 1. reserve_inventory starts running
            # 2. it does successfully reserve inventory
            # 3. but then fails for some other reason (timeout, reporting metrics, etc.)
            # 4. in that case, the activity would have failed, but we still did the effect of reserving inventory
            # So, we need to make sure we have a compensation already on the stack to handle that.
            # This means the compensation needs to handle both the cases of reserved or unreserved inventory.
            compensations.append(lambda: workflow.execute_activity(
                release_inventory_if_reserved, order,
                start_to_close_timeout=timedelta(minutes=5)
            ))
            await workflow.execute_activity(
                reserve_inventory, order,
                start_to_close_timeout=timedelta(minutes=5)
            )

            compensations.append(lambda: workflow.execute_activity(
                refund_payment_if_charged, order,
                start_to_close_timeout=timedelta(minutes=5)
            ))
            await workflow.execute_activity(
                charge_payment, order,
                start_to_close_timeout=timedelta(minutes=5)
            )

            await workflow.execute_activity(
                ship_order, order,
                start_to_close_timeout=timedelta(minutes=5)
            )

            return "Order completed"

        except Exception as e:
            workflow.logger.error(f"Order failed: {e}, running compensations")
            # asyncio.shield ensures compensations run even if the workflow is cancelled.
            async def run_compensations():
                for compensate in reversed(compensations):
                    try:
                        await compensate()
                    except Exception as comp_err:
                        workflow.logger.error(f"Compensation failed: {comp_err}")
            await asyncio.shield(asyncio.ensure_future(run_compensations()))
            raise

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        try:
            await workflow.execute_activity(
                long_running_activity,
                start_to_close_timeout=timedelta(hours=1),
            )
            return "completed"
        except asyncio.CancelledError:
            # Workflow was cancelled - perform cleanup
            workflow.logger.info("Workflow cancelled, running cleanup")
            # Cleanup activities still run even after cancellation
            await workflow.execute_activity(
                cleanup_activity,
                start_to_close_timeout=timedelta(minutes=5),
            )
            raise  # Re-raise to mark workflow as cancelled

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        self._approved = False

        # Wait for approval with 24-hour timeout
        try:
            await workflow.wait_condition(
                lambda: self._approved,
                timeout=timedelta(hours=24)
            )
            return "approved"
        except asyncio.TimeoutError:
            return "auto-rejected due to timeout"

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        # ... main workflow logic ...

        # Before exiting, wait for all handlers to finish
        await workflow.wait_condition(workflow.all_handlers_finished)
        return "done"

from temporalio.exceptions import CancelledError

@activity.defn
def process_large_file(file_path: str) -> str:
    # Get heartbeat details from previous attempt (if any)
    heartbeat_details = activity.info().heartbeat_details
    start_line = heartbeat_details[0] if heartbeat_details else 0

    try:
        with open(file_path) as f:
            for i, line in enumerate(f):
                if i < start_line:
                    continue  # Skip already processed lines

                process_line(line)

                # Heartbeat with progress
                # If cancelled, heartbeat() raises CancelledError
                activity.heartbeat(i + 1)

        return "completed"
    except CancelledError:
        # Perform cleanup on cancellation
        cleanup()
        raise

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        await workflow.sleep(timedelta(hours=1))

        return "Timer fired"

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> str:
        result = await workflow.execute_local_activity(
            quick_lookup,
            "key",
            start_to_close_timeout=timedelta(seconds=5),
        )
        return result