zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增任务跳过原因记录，完善上游任务跳过传播逻辑

Browse Source 
1. 为TaskResult和TaskEvent新增reason字段记录跳过原因
2. 为同步/异步任务执行器添加上游任务跳过检测，自动跳过下游任务
3. 完善任务跳过的原因判断，支持条件不满足、缓存命中、上游跳过场景
4. 优化gittool工具，新增排除目录配置和更灵活的git操作流程
5. 重构测试用例格式，新增上游任务跳过的测试覆盖
6. 默认启用verbose输出，优化跳过任务的日志提示
This commit is contained in:
gooker_young
2026-06-21 21:43:55 +08:00
parent 2f3041c169
commit 9cc91d1153
4 changed files with 249 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/cli/gittool.py
+57 -9
View File
@@ -10,39 +10,87 @@ from pathlib import Path
import pyflowx as px
EXCLUDE_DIRS = [
# 编辑器相关目录
".vscode",
".idea",
".editorconfig",
".trae",
".qoder",
# 项目相关目录
".venv",
".git",
".tox",
"node_modules",
]
EXCLUDE_CMDS = [arg for d in EXCLUDE_DIRS for arg in ["-e", d]]
def init_sub_dirs() -> None:
"""初始化子目录的Git仓库."""
sub_dirs = [subdir for subdir in Path.cwd().iterdir() if subdir.is_dir()]
for subdir in sub_dirs:
px.run(
px.Graph.from_specs(
[
px.TaskSpec("init", cmd=["git", "init"], cwd=str(subdir)),
px.TaskSpec("add", cmd=["git", "add", "."], depends_on=["init"], cwd=str(subdir)),
px.Graph.from_specs([
px.TaskSpec(
"commit", cmd=["git", "commit", "-m", "init commit"], depends_on=["add"], cwd=str(subdir)
"init",
cmd=["git", "init"],
conditions=[not_has_git_repo],
cwd=str(subdir),
),
]
),
verbose=True,
px.TaskSpec("add", cmd=["git", "add", "."], depends_on=["init"], cwd=str(subdir)),
px.TaskSpec("commit", cmd=["git", "commit", "-m", "init commit"], depends_on=["add"], cwd=str(subdir)),
]),
)
isub: px.TaskSpec = px.TaskSpec("isub", fn=init_sub_dirs)
push: px.TaskSpec = px.TaskSpec("push", cmd=["git", "push"])
pull: px.TaskSpec = px.TaskSpec("pull", cmd=["git", "pull"])
kill_tgit: px.TaskSpec = px.TaskSpec("task_kill", cmd=["taskkill", "/f", "/t", "/im", "tgitcache.exe"])
def not_has_git_repo() -> bool:
"""检查当前目录没有Git仓库."""
return not Path.cwd().exists() or not (Path.cwd() / ".git").is_dir()
def has_files() -> bool:
"""检查当前目录是否有文件."""
return bool(list(Path.cwd().glob("*")))
def main() -> None:
"""Git工具主函数."""
runner = px.CliRunner(
strategy="thread",
description="Gittool - Git 执行工具.",
graphs={
"isub": px.Graph.from_specs([px.TaskSpec("isub", fn=init_sub_dirs)]),
# 添加并提交
"a": px.Graph.from_specs([
px.TaskSpec("add", cmd=["git", "add", "."], conditions=[has_files]),
px.TaskSpec("commit", cmd=["git", "commit", "-m", "chore: update"], depends_on=["add"]),
]),
# 清理
"c": px.Graph.from_specs([
px.TaskSpec("clean", cmd=["git", "clean", "-xfd", *EXCLUDE_CMDS]),
px.TaskSpec("status", cmd=["git", "status", "--porcelain"], depends_on=["clean"]),
]),
# 初始化、添加并提交
"i": px.Graph.from_specs([
px.TaskSpec("init", cmd=["git", "init"], conditions=[not_has_git_repo]),
px.TaskSpec("add", cmd=["git", "add", "."], depends_on=["init"], conditions=[has_files]),
px.TaskSpec(
"commit", cmd=["git", "commit", "-m", "init commit"], depends_on=["add"], conditions=[has_files]
),
]),
# 初始化子目录
"isub": px.Graph.from_specs([isub]),
# 推送
"p": px.Graph.from_specs([push]),
# 拉取
"pl": px.Graph.from_specs([pull]),
# 重启TGit缓存
"r": px.Graph.from_specs([kill_tgit]),
},
)
src/pyflowx/executors.py
+60 -8
View File
@@ -54,6 +54,7 @@ def _emit(
attempts=result.attempts,
error=repr(result.error) if result.error else None,
duration=result.duration,
reason=result.reason,
)
)
@@ -91,14 +92,39 @@ def _run_sync_with_retry(
context: Mapping[str, Any],
layer_idx: int | None,
on_event: EventCallback | None = None,
report: RunReport | None = None,
) -> TaskResult[Any]:
"""执行同步任务并带重试；返回填充好的 TaskResult。"""
result: TaskResult[Any] = TaskResult(spec=spec)
# 检查上游任务是否被 SKIPPED
if report is not None:
for dep in spec.depends_on:
if dep in report.results and report.results[dep].status == TaskStatus.SKIPPED:
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = f"上游任务 '{dep}' 被跳过"
logger.info("task %r skipped (上游任务 %r 被跳过)", spec.name, dep)
return result
# 检查条件是否满足
if not spec.should_execute():
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
# 检查是哪个条件不满足
failed_conditions = []
for condition in spec.conditions:
try:
if not condition():
failed_conditions.append(condition.__name__ or "匿名条件")
except Exception:
failed_conditions.append(condition.__name__ or "匿名条件(执行错误)")
if failed_conditions:
result.reason = f"条件不满足: {', '.join(failed_conditions)}"
elif spec.skip_if_missing and not spec._is_cmd_available():
result.reason = f"命令不存在: {spec.cmd[0] if spec.cmd else 'unknown'}"
else:
result.reason = "条件不满足"
logger.info("task %r skipped (条件不满足)", spec.name)
return result
@@ -126,14 +152,39 @@ async def _run_async_with_retry(
context: Mapping[str, Any],
layer_idx: int | None,
on_event: EventCallback | None = None,
report: RunReport | None = None,
) -> TaskResult[Any]:
"""在事件循环上执行任务（同步或异步）并带重试。"""
result: TaskResult[Any] = TaskResult[Any](spec=spec)
# 检查上游任务是否被 SKIPPED
if report is not None:
for dep in spec.depends_on:
if dep in report.results and report.results[dep].status == TaskStatus.SKIPPED:
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = f"上游任务 '{dep}' 被跳过"
logger.info("task %r skipped (上游任务 %r 被跳过)", spec.name, dep)
return result
# 检查条件是否满足
if not spec.should_execute():
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
# 检查是哪个条件不满足
failed_conditions = []
for condition in spec.conditions:
try:
if not condition():
failed_conditions.append(condition.__name__ or "匿名条件")
except Exception:
failed_conditions.append(condition.__name__ or "匿名条件(执行错误)")
if failed_conditions:
result.reason = f"条件不满足: {', '.join(failed_conditions)}"
elif spec.skip_if_missing and not spec._is_cmd_available():
result.reason = f"命令不存在: {spec.cmd[0] if spec.cmd else 'unknown'}"
else:
result.reason = "条件不满足"
logger.info("task %r skipped (条件不满足)", spec.name)
return result
@@ -207,12 +258,12 @@ def _execute_layer_sequential(
if backend.has(name):
cached = backend.get(name)
context[name] = cached
result = TaskResult(spec=spec, status=TaskStatus.SKIPPED, value=cached)
result = TaskResult(spec=spec, status=TaskStatus.SKIPPED, value=cached, reason="缓存命中")
report.results[name] = result
_emit(on_event, result)
logger.info("task %r skipped (cached)", name)
continue
result = _run_sync_with_retry(spec, _build_context(spec, context), layer_idx, on_event)
result = _run_sync_with_retry(spec, _build_context(spec, context), layer_idx, on_event, report)
context[name] = result.value
backend.save(name, result.value)
report.results[name] = result
@@ -236,7 +287,7 @@ def _execute_layer_threaded(
if backend.has(name):
cached = backend.get(name)
context[name] = cached
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached)
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached, reason="缓存命中")
report.results[name] = result
_emit(on_event, result)
else:
@@ -251,7 +302,7 @@ def _execute_layer_threaded(
spec = graph.spec(name)
# 为本任务快照上下文以避免竞态。
task_ctx = _build_context(spec, context)
fut = pool.submit(_run_sync_with_retry, spec, task_ctx, layer_idx, on_event)
fut = pool.submit(_run_sync_with_retry, spec, task_ctx, layer_idx, on_event, report)
future_to_name[fut] = name
for fut in concurrent.futures.as_completed(future_to_name):
@@ -278,7 +329,7 @@ async def _execute_layer_async(
if backend.has(name):
cached = backend.get(name)
context[name] = cached
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached)
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached, reason="缓存命中")
report.results[name] = result
_emit(on_event, result)
else:
@@ -291,7 +342,7 @@ async def _execute_layer_async(
for name in to_run:
spec = graph.spec(name)
task_ctx = _build_context(spec, context)
coros.append(_run_async_with_retry(spec, task_ctx, layer_idx, on_event))
coros.append(_run_async_with_retry(spec, task_ctx, layer_idx, on_event, report))
results = await asyncio.gather(*coros)
for name, result in zip(to_run, results):
@@ -334,7 +385,8 @@ def _make_verbose_callback(
flush=True,
)
elif event.status == TaskStatus.SKIPPED: # pragma: no branch
print(f"[verbose] 任务 {event.task!r} 跳过", flush=True)
reason = f" ({event.reason})" if event.reason else ""
print(f"[verbose] 任务 {event.task!r} 跳过{reason}", flush=True)
else: # pragma: no cover
# 不可达: 执行器只发出 RUNNING/SUCCESS/FAILED/SKIPPED 事件
pass
@@ -351,7 +403,7 @@ def run(
*,
max_workers: int | None = None,
dry_run: bool = False,
verbose: bool = False,
verbose: bool = True,
on_event: EventCallback | None = None,
state: StateBackend | None = None,
) -> RunReport:
src/pyflowx/task.py
+2
View File
@@ -308,6 +308,7 @@ class TaskResult(Generic[T]):
attempts: int = 0
started_at: Optional[datetime] = None
finished_at: Optional[datetime] = None
reason: Optional[str] = None # 跳过原因
@property
def duration(self) -> Optional[float]:
@@ -330,3 +331,4 @@ class TaskEvent:
attempts: int = 0
error: Optional[str] = None
duration: Optional[float] = None
reason: Optional[str] = None # 跳过原因，如 "条件不满足"、"上游任务被跳过"、"缓存"
tests/test_executors.py
+101 -44
View File
@@ -26,12 +26,10 @@ def test_sequential_basic() -> None:
def double(extract: list[int]) -> list[int]:
return [x * 2 for x in extract]
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("extract", extract),
px.TaskSpec("double", double, depends_on=("extract",)),
]
)
])
report = px.run(graph, strategy="sequential")
assert report.success
assert report["extract"] == [1, 2, 3]
@@ -48,14 +46,12 @@ def test_sequential_diamond() -> None:
return fn
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
]
)
])
report = px.run(graph, strategy="sequential")
assert report.success
assert report["d"] == "d"
@@ -69,12 +65,10 @@ def test_failure_propagates() -> None:
def downstream(_boom: None) -> int:
return 1
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("boom", boom),
px.TaskSpec("downstream", downstream, depends_on=("boom",)),
]
)
])
with pytest.raises(TaskFailedError) as exc_info:
_ = px.run(graph, strategy="sequential")
assert exc_info.value.task == "boom"
@@ -116,13 +110,11 @@ def test_threaded_parallelism() -> None:
time.sleep(0.3)
return "done"
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
]
)
])
start = time.time()
report = px.run(graph, strategy="thread", max_workers=3)
elapsed = time.time() - start
@@ -145,13 +137,11 @@ def test_threaded_layer_barrier() -> None:
return fn
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b")),
px.TaskSpec("c", make("c"), depends_on=("a", "b")),
]
)
])
report = px.run(graph, strategy="thread", max_workers=2)
assert report.success
# c must finish after both a and b.
@@ -170,12 +160,10 @@ def test_async_basic() -> None:
async def transform(fetch: int) -> int:
return fetch * 2
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("fetch", fetch),
px.TaskSpec("transform", transform, depends_on=("fetch",)),
]
)
])
report = px.run(graph, strategy="async")
assert report.success
assert report["transform"] == 84
@@ -187,13 +175,11 @@ def test_async_parallelism() -> None:
await asyncio.sleep(0.3)
return "done"
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", slow),
px.TaskSpec("b", slow),
px.TaskSpec("c", slow),
]
)
])
start = time.time()
report = px.run(graph, strategy="async")
elapsed = time.time() - start
@@ -209,12 +195,10 @@ def test_async_mixed_sync_and_async() -> None:
await asyncio.sleep(0.01)
return sync_task + 5
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("sync_task", sync_task),
px.TaskSpec("async_task", async_task, depends_on=("sync_task",)),
]
)
])
report = px.run(graph, strategy="async")
assert report.success
assert report["async_task"] == 15
@@ -262,12 +246,10 @@ def test_memory_backend_resume() -> None:
return fn
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
]
)
])
backend = MemoryBackend()
_ = px.run(graph, strategy="sequential", state=backend)
assert runs == ["a", "b"]
@@ -393,12 +375,10 @@ def test_threaded_skips_cached_tasks() -> None:
return fn
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
]
)
])
backend = px.MemoryBackend()
# 第一次运行填充缓存
_ = px.run(graph, strategy="thread", max_workers=2, state=backend)
@@ -438,12 +418,10 @@ def test_async_skips_cached_tasks() -> None:
runs.append("b")
return a + "b"
graph = px.Graph.from_specs(
[
graph = px.Graph.from_specs([
px.TaskSpec("a", a),
px.TaskSpec("b", b, depends_on=("a",)),
]
)
])
backend = px.MemoryBackend()
_ = px.run(graph, strategy="async", state=backend)
assert runs == ["a", "b"]
@@ -507,3 +485,82 @@ def test_run_empty_graph() -> None:
report = px.run(graph, strategy="sequential")
assert report.success
assert len(report) == 0
# ---------------------------------------------------------------------- #
# 上游任务被 SKIPPED 后，下游任务也应被 SKIPPED
# ---------------------------------------------------------------------- #
def test_downstream_skipped_when_upstream_skipped_sequential() -> None:
"""上游任务被 SKIPPED 后，下游任务也应被 SKIPPED（sequential 策略）."""
never_true = lambda: False # noqa: E731
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
report = px.run(graph, strategy="sequential")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
assert report.result_of("downstream").status == px.TaskStatus.SKIPPED
def test_downstream_skipped_when_upstream_skipped_thread() -> None:
"""上游任务被 SKIPPED 后，下游任务也应被 SKIPPED（thread 策略）."""
never_true = lambda: False # noqa: E731
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", cmd=["echo", "hello"], conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
report = px.run(graph, strategy="thread", max_workers=2)
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
assert report.result_of("downstream").status == px.TaskStatus.SKIPPED
def test_downstream_skipped_when_upstream_skipped_async() -> None:
"""上游任务被 SKIPPED 后，下游任务也应被 SKIPPED（async 策略）."""
async def upstream() -> str:
return "hello"
async def downstream(upstream: str) -> str:
return upstream + "_processed"
never_true = lambda: False # noqa: E731
graph = px.Graph.from_specs([
px.TaskSpec("upstream", upstream, conditions=(never_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
report = px.run(graph, strategy="async")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SKIPPED
assert report.result_of("downstream").status == px.TaskStatus.SKIPPED
def test_downstream_executes_when_upstream_succeeds() -> None:
"""上游任务成功时，下游任务应正常执行."""
always_true = lambda: True # noqa: E731
def upstream() -> str:
return "hello"
def downstream(upstream: str) -> str:
return upstream + "_processed"
graph = px.Graph.from_specs([
px.TaskSpec("upstream", upstream, conditions=(always_true,)),
px.TaskSpec("downstream", downstream, depends_on=("upstream",)),
])
report = px.run(graph, strategy="sequential")
assert report.success
assert report.result_of("upstream").status == px.TaskStatus.SUCCESS
assert report.result_of("downstream").status == px.TaskStatus.SUCCESS
assert report["downstream"] == "hello_processed"