48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
|
"""给 CLI 用的任务模块。
|
||
|
|
|
||
|
|
CLI **只传任务名**，定义必须来自这样一个模块——一个能启动任意可调用对象的命令行
|
||
|
|
入口等于任意代码执行，所以"任务定义从哪里来"必须先有答案。约定就是本文件里的
|
||
|
|
``register(controller)``。
|
||
|
|
|
||
|
|
用法（把本目录加进 ``PYTHONPATH``，或把模块放进你自己的包里）::
|
||
|
|
|
||
|
|
chongming-task --db /tmp/tasks.db tasks --tasks tasks_demo
|
||
|
|
chongming-task --db /tmp/tasks.db run --tasks tasks_demo add --arg left=2 --arg right=3
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import time
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
def sync_tenant(ctx: Any, tenant: str, batches: int = 120) -> str:
|
||
|
|
"""同步一个租户的数据。"""
|
||
|
|
for index in range(batches):
|
||
|
|
ctx.checkpoint()
|
||
|
|
if index % 20 == 0:
|
||
|
|
time.sleep(0.05)
|
||
|
|
return f"{tenant} 同步完成（{batches} 批）"
|
||
|
|
|
||
|
|
|
||
|
|
def add(ctx: Any, left: int, right: int) -> int:
|
||
|
|
"""两数相加。"""
|
||
|
|
del ctx
|
||
|
|
return left + right
|
||
|
|
|
||
|
|
|
||
|
|
def flaky(ctx: Any, fail_times: int = 2) -> str:
|
||
|
|
"""前几次故意失败，用来演示重试。"""
|
||
|
|
state = getattr(flaky, "_attempts", 0) + 1
|
||
|
|
flaky._attempts = state # type: ignore[attr-defined]
|
||
|
|
if state <= fail_times:
|
||
|
|
raise RuntimeError(f"第 {state} 次尝试遇到暂时性故障")
|
||
|
|
return f"第 {state} 次尝试成功"
|
||
|
|
|
||
|
|
|
||
|
|
def register(controller: Any) -> None:
|
||
|
|
"""把任务登记进白名单。CLI 导入本模块后就会调用它。"""
|
||
|
|
controller.register("sync", sync_tenant, tags={"io"})
|
||
|
|
controller.register("add", add, tags={"cpu"})
|
||
|
|
controller.register("flaky", flaky)
|