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

feat: 添加上游任务跳过豁免、进程检查条件及相关优化

Browse Source 
1. 新增allow_upstream_skip参数支持任务不跟随上游跳过
2. 新增IS_RUNNING内置条件检查进程运行状态
3. 调整skip_if_missing默认值为False
4. 补充跳过任务的事件上报和verbose打印
5. 优化reset_icon_cache示例任务使用新特性
6. 更新测试用例匹配默认参数变更
This commit is contained in:
gooker_young
2026-06-27 09:24:22 +08:00
parent a98eb6e344
commit 20c4fb87c5
5 changed files with 87 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/conditions.py
+38
View File
@@ -8,6 +8,7 @@ from __future__ import annotations
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Callable
@@ -66,6 +67,43 @@ class BuiltinConditions:
"""
return sys.version_info >= (major, minor)
@staticmethod
def IS_RUNNING(app_name: str) -> Condition:
"""检查指定应用是否正在运行.
Parameters
----------
app_name : str
应用名称 (如 "explorer", "chrome", "python").
Returns
-------
Condition
条件判断函数.
"""
def _check() -> bool:
if Constants.IS_WINDOWS:
# Windows: 使用 tasklist 命令
result = subprocess.run(
["tasklist", "/nh", "/fi", f"imagename eq {app_name}"],
capture_output=True,
text=True,
check=False,
)
return app_name.lower() in result.stdout.lower()
else:
# Linux/macOS: 使用 pgrep 命令
result = subprocess.run(
["pgrep", "-x", app_name],
capture_output=True,
check=False,
)
return result.returncode == 0
_check.__name__ = f"IS_RUNNING({app_name!r})"
return _check
@staticmethod
def HAS_INSTALLED(app_name: str) -> Condition:
"""检查指定应用是否已安装.
src/pyflowx/executors.py
+16
View File
@@ -101,6 +101,10 @@ def _check_upstream_skipped(
if report is None:
return False, None
# 若任务允许上游跳过，则不检查上游状态
if spec.allow_upstream_skip:
return False, None
for dep in spec.depends_on:
if dep in report.results and report.results[dep].status == TaskStatus.SKIPPED:
return True, f"上游任务 '{dep}' 被跳过"
@@ -155,6 +159,9 @@ def _run_sync_with_retry(
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = skip_reason
_emit(on_event, result)
if spec.verbose:
print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True)
logger.info("task %r skipped (上游任务被跳过)", spec.name)
return result
@@ -164,6 +171,9 @@ def _run_sync_with_retry(
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = skip_reason
_emit(on_event, result)
if spec.verbose:
print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True)
logger.info("task %r skipped (条件不满足)", spec.name)
return result
@@ -232,6 +242,9 @@ async def _run_async_with_retry(
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = skip_reason
_emit(on_event, result)
if spec.verbose:
print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True)
logger.info("task %r skipped (上游任务被跳过)", spec.name)
return result
@@ -241,6 +254,9 @@ async def _run_async_with_retry(
result.status = TaskStatus.SKIPPED
result.finished_at = datetime.now()
result.reason = skip_reason
_emit(on_event, result)
if spec.verbose:
print(f"[skip] 任务 '{spec.name}' 跳过: {skip_reason}", flush=True)
logger.info("task %r skipped (条件不满足)", spec.name)
return result
src/pyflowx/task.py
+6 -1
View File
@@ -123,6 +123,10 @@ class TaskSpec(Generic[T]):
（标记为 SKIPPED）而非失败。适用于构建工具场景，避免因未安装
某些工具（如 maturin、tox）而导致整个图执行失败。
对于 ``str`` (shell) 和 ``Callable`` 类型的 ``cmd``，此参数无效。
allow_upstream_skip:
若为 ``True``，当上游任务因条件不满足被跳过时，本任务仍会执行
（而非跟随跳过）。适用于清理类任务：即使某些删除操作因目标不存在
而跳过，后续操作（如重启服务）仍应执行。默认为 ``False``。
"""
name: str
@@ -137,7 +141,8 @@ class TaskSpec(Generic[T]):
conditions: Tuple[Condition, ...] = ()
cwd: Optional[Path] = None
verbose: bool = False
skip_if_missing: bool = True
skip_if_missing: bool = False
allow_upstream_skip: bool = False
def __post_init__(self) -> None:
if not self.name:
src/pyflowx/tasks/system.py
+19 -9
View File
@@ -27,30 +27,40 @@ def reset_icon_cache() -> list[px.TaskSpec]:
print("reset_icon_cache: 仅在 Windows 上支持")
return []
local_app_data = os.environ.get("LOCALAPPDATA", "")
icon_cache_db = Path(local_app_data) / "IconCache.db"
explorer_cache_dir = Path(local_app_data) / "Microsoft" / "Windows" / "Explorer"
return [
px.TaskSpec(
"kill_explorer",
fn=lambda: subprocess.run(["taskkill", "/f", "/im", "explorer.exe"], check=False),
cmd=["taskkill", "/f", "/im", "explorer.exe"],
conditions=(BuiltinConditions.IS_RUNNING("explorer.exe"),),
verbose=True,
),
px.TaskSpec(
"delete_icon_cache",
fn=lambda: subprocess.run(["del", "/a", "/q", r"%localappdata%\IconCache.db"], check=False),
conditions=(BuiltinConditions.DIR_EXISTS(Path(r"%localappdata%\IconCache.db")),),
cmd=["cmd", "/c", "del", "/a", "/q", str(icon_cache_db)],
conditions=(BuiltinConditions.DIR_EXISTS(icon_cache_db),),
depends_on=("kill_explorer",),
verbose=True,
),
px.TaskSpec(
"delete_icon_cache_all",
fn=lambda: subprocess.run(
["del", "/a", "/q", r"%localappdata%\Microsoft\Windows\Explorer\iconcache*"], check=False
),
conditions=(BuiltinConditions.DIR_EXISTS(Path(r"%localappdata%\Microsoft\Windows\Explorer")),),
cmd=["cmd", "/c", "del", "/a", "/q", str(explorer_cache_dir / "iconcache*")],
conditions=(BuiltinConditions.DIR_EXISTS(explorer_cache_dir),),
depends_on=("kill_explorer",),
verbose=True,
),
px.TaskSpec(
"restart_explorer",
fn=lambda: subprocess.run(["explorer.exe"], check=False),
conditions=(BuiltinConditions.HAS_INSTALLED("explorer.exe"),),
cmd=["cmd", "/c", "start", "explorer.exe"],
conditions=(
BuiltinConditions.HAS_INSTALLED("explorer.exe"),
BuiltinConditions.NOT(BuiltinConditions.IS_RUNNING("explorer.exe")),
),
depends_on=("delete_icon_cache", "delete_icon_cache_all"),
allow_upstream_skip=True,
verbose=True,
),
]
tests/cli/test_pymake.py
+8 -8
View File
@@ -77,25 +77,25 @@ class TestTaskSpecDefinitions:
"""uv_build spec should be properly defined."""
assert pymake.uv_build.name == "uv_build"
assert pymake.uv_build.cmd == ["uv", "build"]
assert pymake.uv_build.skip_if_missing is True
assert pymake.uv_build.skip_if_missing is False
def test_maturin_build_spec(self) -> None:
"""maturin_build spec should be properly defined."""
assert pymake.maturin_build.name == "maturin_build"
assert isinstance(pymake.maturin_build.cmd, list)
assert pymake.maturin_build.skip_if_missing is True
assert pymake.maturin_build.skip_if_missing is False
def test_uv_sync_spec(self) -> None:
"""uv_sync spec should be properly defined."""
assert pymake.uv_sync.name == "uv_sync"
assert pymake.uv_sync.cmd == ["uv", "sync"]
assert pymake.uv_sync.skip_if_missing is True
assert pymake.uv_sync.skip_if_missing is False
def test_git_clean_spec(self) -> None:
"""git_clean spec should be properly defined."""
assert pymake.git_clean.name == "git_clean"
assert pymake.git_clean.cmd == ["gitt", "c"]
assert pymake.git_clean.skip_if_missing is True
assert pymake.git_clean.skip_if_missing is False
def test_test_spec(self) -> None:
"""test spec should be properly defined."""
@@ -135,13 +135,13 @@ class TestTaskSpecDefinitions:
assert pymake.doc.name == "doc"
assert isinstance(pymake.doc.cmd, list)
assert "sphinx-build" in pymake.doc.cmd
assert pymake.doc.skip_if_missing is True
assert pymake.doc.skip_if_missing is False
def test_hatch_publish_spec(self) -> None:
"""hatch_publish spec should be properly defined."""
assert pymake.hatch_publish.name == "publish_python"
assert pymake.hatch_publish.cmd == ["hatch", "publish"]
assert pymake.hatch_publish.skip_if_missing is True
assert pymake.hatch_publish.skip_if_missing is False
def test_twine_publish_spec(self) -> None:
"""twine_publish spec should be properly defined."""
@@ -149,13 +149,13 @@ class TestTaskSpecDefinitions:
assert isinstance(pymake.twine_publish.cmd, list)
assert "twine" in pymake.twine_publish.cmd
assert "upload" in pymake.twine_publish.cmd
assert pymake.twine_publish.skip_if_missing is True
assert pymake.twine_publish.skip_if_missing is False
def test_tox_spec(self) -> None:
"""tox spec should be properly defined."""
assert pymake.tox.name == "tox"
assert pymake.tox.cmd == ["tox", "-p", "auto"]
assert pymake.tox.skip_if_missing is True
assert pymake.tox.skip_if_missing is False
# ---------------------------------------------------------------------- #