请求Prefect 1中DateClocks指定日期任务调度（含时区）代码示例

核心实现思路 #

Prefect 1的DateClocks专门用于指定离散的日期时间触发流程，完全匹配你需要的非cron、非间隔式调度需求。以下是带时区配置的完整代码实现，全程通过代码完成，无需UI操作。

完整代码示例 #

import pytz
from datetime import datetime
from prefect import task, Flow
from prefect.schedules import Schedule
from prefect.schedules.clocks import DateClock

# 1. 定义示例任务（替换为你的实际业务逻辑）
@task
def scheduled_task():
    print(f"任务执行时间：{datetime.now(pytz.timezone('Asia/Shanghai'))}")
    # 这里编写你的任务代码

# 2. 定义流程
with Flow("特定日期触发流程") as flow:
    scheduled_task()

# 3. 配置目标时区（示例为上海时区）
target_timezone = pytz.timezone("Asia/Shanghai")

# 4. 指定需要触发流程的具体日期时间（必须带时区）
trigger_dates = [
    datetime(2024, 6, 15, 10, 0, tzinfo=target_timezone),
    datetime(2024, 7, 20, 14, 30, tzinfo=target_timezone),
    datetime(2024, 8, 5, 9, 15, tzinfo=target_timezone)
]

# 5. 为每个日期创建DateClock实例
clocks = [DateClock(date=dt) for dt in trigger_dates]

# 6. 将调度规则绑定到流程
flow.schedule = Schedule(clocks=clocks)

# 7. 注册流程到Prefect
if __name__ == "__main__":
    flow.register(project_name="你的项目名称")

关键细节说明 #

• 时区处理：通过pytz.timezone()明确指定时区，确保所有触发时间都基于该时区计算，避免跨时区时间偏移问题。
• DateClock创建：每个指定的日期时间对应一个DateClock实例，将这些实例组成列表传入Schedule即可实现多日期触发。
• 流程注册：运行代码后，流程会被注册到指定的Prefect项目中，Prefect代理会自动监听并在目标日期触发任务。

运行前置步骤 #

1. 安装依赖：pip install prefect==1.* pytz
2. 创建Prefect项目（如果没有）：prefect create project "你的项目名称"
3. 运行代码完成流程注册，启动Prefect代理后即可等待任务触发。

内容的提问来源于stack exchange，提问作者atharva narwane