scrubbers 0.1.0

High-throughput redaction engine + CLI
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

#!/usr/bin/env python3
import argparse
import json
import subprocess
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DIST = ROOT / "dist"
SMOKE_TEST = ROOT / "bindings/python/smoke_test.py"


def log(level: str, event: str, detail: str) -> None:
    print(json.dumps({"level": level, "event": event, "detail": detail}))


def run(cmd: list[str], *, verbose: bool) -> None:
    if verbose:
        log("debug", "exec", " ".join(cmd))
        subprocess.run(cmd, cwd=ROOT, check=True)
        return

    result = subprocess.run(
        cmd,
        cwd=ROOT,
        check=False,
        text=True,
        capture_output=True,
    )
    if result.returncode == 0:
        return
    if result.stdout:
        print(result.stdout, end="")
    if result.stderr:
        print(result.stderr, end="", file=sys.stderr)
    raise subprocess.CalledProcessError(result.returncode, cmd)


def latest_artifacts(pattern: str) -> list[Path]:
    return sorted(DIST.glob(pattern), key=lambda path: path.stat().st_mtime_ns, reverse=True)


def build_distributions(kind: str, verbose: bool) -> None:
    if kind == "all":
        cmd = ["uv", "build"]
    else:
        cmd = ["uv", "build", f"--{kind}"]
    log("info", "build", " ".join(cmd))
    run(cmd, verbose=verbose)


def smoke_test(artifact: Path, python: str | None, verbose: bool) -> None:
    cmd = ["uv", "run"]
    if python:
        cmd.extend(["--python", python])
    cmd.extend(
        [
            "--isolated",
            "--no-project",
            "--with",
            str(artifact),
            str(SMOKE_TEST),
        ]
    )
    log("info", "smoke", f"testing {artifact.name}")
    run(cmd, verbose=verbose)


def verify_artifact(kind: str, python: str | None, verbose: bool) -> None:
    pattern = "*.whl" if kind == "wheel" else "*.tar.gz"
    artifacts = latest_artifacts(pattern)
    if not artifacts:
        raise FileNotFoundError(f"no {kind} artifacts found in {DIST}")
    smoke_test(artifacts[0], python, verbose)


def main() -> None:
    parser = argparse.ArgumentParser(description="Build and smoke test the Python package distributions.")
    parser.add_argument(
        "--artifact",
        choices=["wheel", "sdist", "all"],
        default="all",
        help="Which distribution kind to build and validate",
    )
    parser.add_argument(
        "--python",
        default=None,
        help="Optional Python version or interpreter path to use for uv run smoke tests",
    )
    parser.add_argument(
        "--skip-build",
        action="store_true",
        help="Reuse the current dist/ contents instead of rebuilding",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Stream subprocess output directly",
    )
    args = parser.parse_args()

    if not args.skip_build:
        build_distributions(args.artifact, args.verbose)

    if args.artifact in ("wheel", "all"):
        verify_artifact("wheel", args.python, args.verbose)
    if args.artifact in ("sdist", "all"):
        verify_artifact("sdist", args.python, args.verbose)

    log("info", "done", f"{args.artifact} distribution checks passed")


if __name__ == "__main__":
    main()