调度工作流

工作流调度是自动化流程中的关键环节，在处理对时间敏感的任务时尤为重要。通过调度工作流，可以实现重复性任务的自动化，减少人工干预的需求，确保业务流程的及时执行。

可以使用以下操作调度工作流执行，从而掌控自动化流程。

调度的行为由其重叠策略（Overlap Policy）决定。如果调度启动的工作流执行处于暂停（Paused）状态，该执行将保持开启，并且在进行重叠决策时被计为正在运行中的执行。

创建调度工作流

创建操作用于新建调度。创建新调度时，系统生成唯一的调度 ID，可以在其他调度命令中使用该 ID 引用此调度。

要在 Python 中创建调度工作流执行，需使用客户端的 create_schedule() 异步方法。将调度 ID 和调度对象传递给该方法，即可创建调度工作流执行。将 action 参数设置为 ScheduleActionStartWorkflow ，启动工作流执行。还可以选择将 spec 参数设置为 ScheduleSpec ，指定调度计划，或将 intervals 参数设置为 ScheduleIntervalSpec，指定时间间隔。其他可用选项包括：cron_expressions、skip、start_at 和 jitter。

import asyncio
from datetime import timedelta

from temporalio.client import (
    Client,
    Schedule,
    ScheduleActionStartWorkflow,
    ScheduleIntervalSpec,
    ScheduleSpec,
    ScheduleState,
)
from your_workflows import YourSchedulesWorkflow


async def main():
    client = await Client.connect("localhost:7233")

await client.create_schedule(
"workflow-schedule-id",
        Schedule(
            action=ScheduleActionStartWorkflow(
                YourSchedulesWorkflow.run,
"my schedule arg",
id="schedules-workflow-id",
                task_queue="schedules-task-queue",
            ),
            spec=ScheduleSpec(
                intervals=[ScheduleIntervalSpec(every=timedelta(minutes=2))]
            ),
            state=ScheduleState(note="Here's a note on my Schedule."),
        ),
    )


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

调度自动删除

当调度完成其所有工作流执行的创建后，Temporal 服务会将其删除，因为该调度不再触发新执行。Temporal 服务不保证此删除操作的具体发生时间。

回填调度工作流

回填操作执行规定时间范围之前的操作。当需要执行错过的或延迟的操作，或者希望测试调度时间之前的工作流时，此命令非常有用。

要在 Python 中对调度工作流执行进行回填，需要使用调度句柄的 backfill() 异步方法。

import asyncio
from datetime import datetime, timedelta

from temporalio.client import Client, ScheduleBackfill, ScheduleOverlapPolicy

async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )
    now = datetime.utcnow()
    (
await handle.backfill(
            ScheduleBackfill(
                start_at=now - timedelta(minutes=10),
                end_at=now - timedelta(minutes=9),
                overlap=ScheduleOverlapPolicy.ALLOW_ALL,
            ),
        ),
    )

    print(f"Result: {handle}")


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

删除调度工作流

删除操作用于删除调度。删除调度不会影响该调度已启动的工作流。

要在 Python 中删除调度工作流执行，需要使用调度句柄的 delete() 异步方法。

import asyncio

from temporalio.client import Client


async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )

await handle.delete()


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

描述调度工作流

描述操作用于查看当前调度的配置，包括过去、当前和未来的工作流运行（Workflow Run）的相关信息。当需要获取调度及其关联工作流运行的详细视图时，此命令非常有用。

要在 Python 中描述调度工作流执行，需要使用调度句柄的 describe() 异步方法。可以从 ScheduleDescription 类中获取调度工作流执行的完整属性列表。

import asyncio

from temporalio.client import Client


async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )

    desc =await handle.describe()

print(f"Returns the note:{desc.schedule.state.note}")


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

列出调度工作流

列表操作用于列出所有可用的调度。当需要查看所有调度及其对应的调度 ID 时，此命令非常有用。

要列出所有调度，需要使用客户端的 list_schedules() 异步方法。如果添加或删除调度，这些调度无法立即在列表中反映出来。

import asyncio

from temporalio.client import Client


async def main() -> None:
    client = await Client.connect("localhost:7233")

asyncfor scheduleinawait client.list_schedules():
        print(f"List Schedule Info: {schedule.info}.")


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

暂停调度工作流

暂停操作用于暂停和恢复调度。暂停调度时，与该调度关联的所有未来工作流运行都将被暂时停止。当因维护或其他原因需要暂时中止工作流时，此命令非常有用。

要在 Python 中暂停调度工作流执行，需要使用调度句柄的 pause() 异步方法。可以通过向 pause() 方法传入备注，说明暂停调度的原因。

import asyncio

from temporalio.client import Client


async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )

await handle.pause(note="Pausing the schedule for now")


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

触发调度工作流

触发操作用于立即执行与给定调度对应的操作。默认情况下，此操作受调度的重叠策略约束。当需要在调度时间之外执行工作流时，此命令非常有用。

要在 Python 中触发调度工作流执行，需要使用调度句柄的 trigger() 异步方法。

import asyncio

from temporalio.client import Client


async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )

await handle.trigger()


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

更新调度工作流

更新操作用于更新现有调度。当需要修改调度的配置（比如变更开始时间、结束时间或时间间隔）时，此命令非常有用。

创建接收 ScheduleUpdateInput，返回 ScheduleUpdate 的函数。若要更新调度，则需要使用回调函数根据当前描述，构建更新。下面的示例使用新参数更新调度。

import asyncio

from temporalio.client import (
    Client,
    ScheduleActionStartWorkflow,
    ScheduleUpdate,
    ScheduleUpdateInput,
)


async def main():
    client = await Client.connect("localhost:7233")
    handle = client.get_schedule_handle(
        "workflow-schedule-id",
    )

    async def update_schedule_simple(input: ScheduleUpdateInput) -> ScheduleUpdate:
        schedule_action = input.description.schedule.action

        if isinstance(schedule_action, ScheduleActionStartWorkflow):
            schedule_action.args = ["my new schedule arg"]
        return ScheduleUpdate(schedule=input.description.schedule)

await handle.update(update_schedule_simple)


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

延迟开始

使用 start_delay 可以在未来的特定时间点调度一次性的工作流执行。

在客户端的 start_workflow() 或 execute_workflow() 异步方法中使用 start_delay 选项。

async def main():
    client = await Client.connect("localhost:7233")

    result = await client.execute_workflow(
        YourWorkflow.run,
        "your name",
        id="your-workflow-id",
        task_queue="your-task-queue",
        start_delay=timedelta(hours=1, minutes=20, seconds=30)
    )

    print(f"Result: {result}")


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