processkit 3.2.0

Async child-process management for tokio: whole-tree kill-on-drop (no orphans), plus streaming, pipelines, timeouts, and supervision
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

#!/usr/bin/env python3
"""Remove disposable local build caches without deleting target roots."""

from __future__ import annotations

import os
from pathlib import Path
import shutil
import sys


REPOSITORY = Path(__file__).resolve().parent.parent


def is_link(path: Path) -> bool:
    """Return whether deleting the path could traverse outside its parent."""

    is_junction = getattr(path, "is_junction", lambda: False)
    return path.is_symlink() or is_junction()


def incremental_directories(target: Path) -> list[Path]:
    """Find incremental roots without following nested links or junctions."""

    found: list[Path] = []
    for current, directories, _ in os.walk(target, followlinks=False):
        current_path = Path(current)
        retained: list[str] = []
        for name in directories:
            candidate = current_path / name
            if is_link(candidate):
                continue
            if name == "incremental":
                found.append(candidate)
            else:
                retained.append(name)
        directories[:] = retained
    return found


def target_roots() -> list[Path]:
    roots = [REPOSITORY / "target"]
    worktrees = REPOSITORY / ".work" / "worktrees"
    if worktrees.is_dir():
        roots.extend(path / "target" for path in worktrees.iterdir() if path.is_dir())
    return [root for root in roots if root.is_dir()]


def remove_tree(path: Path) -> bool:
    if is_link(path):
        print(f"skip linked path: {path}", file=sys.stderr)
        return False
    shutil.rmtree(path)
    print(f"removed {path}")
    return True


def main() -> int:
    removed = 0
    for target in target_roots():
        for incremental in incremental_directories(target):
            removed += remove_tree(incremental)

        checkout = target.parent
        for name in ("mutants.out", "mutants.out.old"):
            report = checkout / name
            if report.is_dir():
                removed += remove_tree(report)

    suffix = "y" if removed == 1 else "ies"
    print(f"clean-disk removed {removed} disposable director{suffix}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())