temporal tutorial python

Guri Bee

3 min read

·

20-10-2024

1

0

Share

Temporal Tutorial: Building Robust, Scalable Python Workflows

Temporal is an open-source workflow orchestration engine that helps you build robust and scalable applications. Its key strength lies in its ability to manage complex, stateful workflows that span multiple steps and involve different services. In this tutorial, we'll dive into the world of Temporal using Python, covering the basics of workflow and activity definitions, scheduling, and more.

Why Choose Temporal?

Temporal shines when you're dealing with workflows that are:

• Long-running: Tasks that take minutes, hours, or even days to complete.
• Distributed: Involving multiple services or systems.
• Fault-tolerant: Resilient to failures and able to recover gracefully.

For example, you might use Temporal for:

• Order processing: A workflow that handles order placement, payment verification, inventory management, and shipping updates.
• Data processing pipelines: A workflow that pulls data from various sources, transforms it, and loads it into a data warehouse.
• Machine learning pipelines: A workflow that trains a machine learning model, evaluates its performance, and deploys it.

Getting Started with Temporal and Python

Before we begin, make sure you have Temporal installed. Follow the installation guide for your operating system on the Temporal website.

1. Basic Workflow and Activity Definitions

from temporalio import workflow, activity
from temporalio.worker import Worker

# Define an activity
@activity.defn
async def greet(name: str) -> str:
    return f"Hello, {name}!"

# Define a workflow
@workflow.defn
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        greeting = await self.greet(name)
        return f"Workflow completed: {greeting}"

    @workflow.method
    async def greet(self, name: str) -> str:
        return await self.activity_stub.greet(name)

# Create a Temporal worker
worker = Worker(
    task_queue="greet-queue",
    workflows=[GreetingWorkflow],
    activities=[greet],
)
worker.run()

Explanation:

• Activities: Represent individual tasks within a workflow. In this example, greet is an activity that takes a name and returns a greeting.
• Workflows: Define the overall flow of activities. GreetingWorkflow orchestrates the greet activity, handling scheduling and potential retries.
• Workflow Methods: Allow you to define methods within a workflow that can call activities and other methods.

2. Scheduling and Handling Results

@workflow.defn
class OrderProcessingWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        # Place order
        await self.place_order(order_id)
        # Verify payment
        await self.verify_payment(order_id)
        # Ship order
        await self.ship_order(order_id)
        return f"Order {order_id} processed successfully!"

    @workflow.method
    async def place_order(self, order_id: str) -> None:
        await self.activity_stub.place_order(order_id)

    @workflow.method
    async def verify_payment(self, order_id: str) -> None:
        await self.activity_stub.verify_payment(order_id)

    @workflow.method
    async def ship_order(self, order_id: str) -> None:
        await self.activity_stub.ship_order(order_id)

Explanation:

• The OrderProcessingWorkflow defines a workflow for processing an order.
• It calls three activities (place_order, verify_payment, ship_order) in a sequential manner.
• The activity_stub provides access to the activities from within the workflow.

3. Error Handling and Retries

Temporal handles failures and retries gracefully. You can control retry policies for activities using decorators.

@activity.defn(
    schedule_to_close_timeout=60,  # Activity timeout
    retry_options={
        "maximum_attempts": 3,  # Maximum retry attempts
        "initial_interval": 1,  # Initial retry interval in seconds
    }
)
async def place_order(order_id: str) -> None:
    # ... implementation ...

Explanation:

• The schedule_to_close_timeout specifies the maximum time an activity can run.
• The retry_options define retry parameters like maximum attempts and initial interval.

4. Timeouts and Deadlines

Temporal provides mechanisms for handling timeouts and setting deadlines for workflows and activities.

@workflow.defn(
    task_queue="order-queue",
    execution_timeout=3600,  # Workflow execution timeout
)
class OrderProcessingWorkflow:
    # ...

Explanation:

• The execution_timeout defines the maximum time a workflow can run before it's considered timed out.

5. State Management and Persistence

Temporal handles state persistence automatically, so you don't need to worry about managing database interactions.

6. Debugging and Monitoring

Temporal offers powerful debugging and monitoring tools, including:

• Temporal Web UI: Provides real-time insights into workflow and activity execution.
• Temporal CLI: Allows you to interact with the Temporal cluster.
• Logs and Metrics: Capture valuable data for troubleshooting and performance analysis.

Conclusion

Temporal offers a powerful and flexible framework for building robust, scalable workflows with Python. By leveraging its features, you can simplify complex orchestration tasks and improve the resilience and reliability of your applications.

This tutorial provides a basic introduction. Further exploration into Temporal's documentation, tutorials, and community resources can unlock the full potential of this valuable tool.