chakcy
f4de43c743
单任务受控执行（15 态状态机 + 审计轨迹）、DAG 管道、Controller 门面。
执行器三个：线程（默认，含 async def 任务）、内联（同步 / 测试）、进程
（spawn 子进程：硬性停止、绕开 GIL、崩溃隔离）。
调度：并发额度 + 每桶限流 + 具名资源信号量（按标签）+ 优先级排队。
持久层：TaskStore 契约 + 内存 / SQLite / Postgres 实现，契约测试随包发布。
跨进程与多节点：写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。
门禁：ruff / ruff format / mypy(strict) 干净；
pytest 563 passed（44 skipped 是需要真实 Postgres 的那批）；
带一台可连的 Postgres 时 607 passed。
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)
|