3 - 快速入门 - Python 版

配置本地开发环境，开始使用 Temporal 进行开发。

安装 Python

确保已安装 Python，使用如下命令检查 Python 版本：

python3 -V

安装 Temporal Python SDK

在虚拟环境中安装 Temporal Python SDK。为 Temporal 项目创建新目录，切换到新目录，创建 Python 虚拟环境，激活虚拟环境，然后安装 Temporal SDK：

mkdir temporal-project
cd temporal-project
python3 -m venv env
source env/bin/activate
pip install temporalio

下面配置用于开发的本地 Temporal 服务。

安装 Temporal CLI

在本地机器上运行开发版 Temporal 服务最快的方式是使用 Temporal CLI。

下面以 Linux 为例进行说明：

  1. 下载 Temporal CLI：
    1. Linux amd64
    1. Linux arm64
  1. 解压归档文件，将 temporal 二进制文件移动到 PATH 中，比如：
    sudo mv temporal /usr/local/bin

启动开发服务

安装 Temporal CLI，将其添加到 PATH 后，打开新终端窗口，运行以下命令：

temporal server start-dev --ip 0.0.0.0

该命令启动本地 Temporal 服务。它启动 Web UI、创建默认命名空间（Namespace），使用内存数据库。

Temporal 服务的地址是 localhost:7233，Temporal Web UI 的地址是 http://localhost:8233。

如需停止服务，随时按 CTRL+C 即可。

一切安装就绪后，即可在本地机器上使用 Temporal 构建应用程序。

更改 Web UI 端口

在某些示例或教程中，Temporal Web UI 可能使用不同的端口。如需更改 Web UI 的端口，可在启动服务器时使用 --ui-port 选项：

temporal server start-dev --ui-port 8080

运行 Hello World：测试安装

通过创建及运行包含工作流和 Activity 的完整 Temporal 应用程序，验证环境配置是否正常工作。

本次测试将确认以下内容：

创建 Activity

创建 Activity 文件（activities.py）：

from temporalio import activity

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

Activity 是普通函数或方法，用于执行单个定义明确的动作（可以短时间运行，也可以长时间运行），这些动作通常涉及与外部世界的交互，比如发送电子邮件、发起网络请求、写数据库或调用 API，这些操作容易发生故障。如果Activity 失败，Temporal 将根据配置自动重试。

创建工作流

创建工作流文件（workflows.py）：

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities import greet

@workflow.defn
class SayHelloWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await workflow.execute_activity(
            greet,
            name,
            schedule_to_close_timeout=timedelta(seconds=10),
        )

工作流编排 Activity，包含应用程序逻辑。Temporal 工作流具有弹性。即便底层基础设施出现故障，它们也能持续运行，甚至长达数年之久。如果应用程序自身崩溃，Temporal 将自动重建其故障前的状态，使其能够从断点处继续执行。

创建 Worker

创建 Worker 文件（worker.py）：

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from workflows import SayHelloWorkflow
    from activities import greet

async def main():
    client = await Client.connect("localhost:7233")
    worker = Worker(
        client,
        task_queue="my-task-queue",
        workflows=[SayHelloWorkflow],
        activities=[greet],
    )
    print("Worker started.")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

打开新终端，运行 Worker：

source env/bin/activate
python3 worker.py

保持此终端运行——应该看到“Worker started”。

在定义好 Activity 和工作流后，需要 Worker 来执行它们。Worker 轮询为其配置的任务队列（Task Queue），寻找待处理任务。Worker 从任务队列中取出 Workflow 或 Activity 任务后，将执行该任务。

执行工作流

现在开始启动工作流执行。该步骤将验证一切是否正常运行。创建名为 starter.py 的独立文件：

import asyncio
import uuid
from temporalio.client import Client

async def main():
    client = await Client.connect("localhost:7233")
    result = await client.execute_workflow(
        "SayHelloWorkflow",
        "Temporal",
        id=f"say-hello-workflow-{uuid.uuid4()}",
        task_queue="my-task-queue",
    )
    print("Workflow result:", result)

if __name__ == "__main__":
    asyncio.run(main())

在新终端中，运行如下命令：

source env/bin/activate
python3 starter.py

验证成功

如果一切正常，将看到：