daylin/swydd
1
0
Fork 0
mirror of https://github.com/daylinmorgan/swydd.git synced 2024-11-23 15:20:43 -06:00
Code Issues Projects Releases Packages Wiki Activity

refactor!: drop operators

Browse source
This commit is contained in:
Daylin Morgan 2024-10-22 11:53:19 -05:00
parent 115900bf83
commit 277f7f7c0c
Signed by: daylin
GPG key ID: 950D13E9719334AD
3 changed files with 97 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

102
src/swydd/__init__.py
View file

@ -134,7 +134,9 @@ class Context:
for task in self._tasks.values():
if not task.targets:
continue
print(task.name)
print(task.targets)
print(task.needs)
for target in task.targets:
if not task.needs:
self._graph.add_nodes(task, target, None)
@ -151,7 +153,7 @@ class Context:
ctx = Context()
def define_env(key: str, value: str) -> None:
def setenv(key: str, value: str) -> None:
ctx._env.update({key: value})
@ -183,18 +185,6 @@ class SwyddSubResult:
class SwyddProc:
"""
usage:
sub < proc("echo $WORD world", env={'WORD':'hello'})
sub < (proc | "single proc")
sub < (proc | "echo hello" | "wc -c")
sub(proc("hello world").pipe("wc -c"))
sub < (proc & "cat -a" & "echo unreachable")
sub(proc("cat -a").then("echo unreachable")
"""
def __init__(
self, cmd: str | None = None, output: bool = False, **kwargs: Any
) -> None:
@ -204,6 +194,10 @@ class SwyddProc:
self.output = output
self.cmd_kwargs = kwargs
@classmethod
def __call__(cls, *args, **kwargs) -> "SwyddProc":
return cls(*args, **kwargs)
def pipe(self, proc: "str | SwyddProc") -> "SwyddPipe | SwyddProc":
if isinstance(proc, str):
if self._cmd is None:
@ -213,8 +207,8 @@ class SwyddProc:
elif isinstance(proc, SwyddProc):
return SwyddPipe(proc)
def __or__(self, proc: "str | SwyddProc") -> "SwyddPipe | SwyddProc":
return self.pipe(proc)
# def __or__(self, proc: "str | SwyddProc") -> "SwyddPipe | SwyddProc":
# return self.pipe(proc)
def then(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
if self._cmd:
@ -228,8 +222,8 @@ class SwyddProc:
else:
return SwyddSeq(SwyddProc(proc))
def __and__(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
return self.then(proc)
# def __and__(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
# return self.then(proc)
def _build_kwargs(self) -> Dict[str, Any]:
sub_kwargs: Dict[str, Any] = dict(env={**os.environ, **ctx._env})
@ -297,8 +291,8 @@ class SwyddPipe:
def pipe(self, proc: "str | SwyddProc | SwyddPipe") -> "SwyddPipe":
return SwyddPipe(self, proc)
def __or__(self, proc: "str | SwyddProc | SwyddPipe") -> "SwyddPipe":
return self.pipe(proc)
# def __or__(self, proc: "str | SwyddProc | SwyddPipe") -> "SwyddPipe":
# return self.pipe(proc)
class SwyddSeq:
@ -319,8 +313,8 @@ class SwyddSeq:
def then(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
return SwyddSeq(*self._procs, proc)
def __and__(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
return self.then(proc)
# def __and__(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq":
# return self.then(proc)
def execute(self, output: bool = False) -> "SwyddSubResult":
results = []
@ -342,6 +336,7 @@ class SwyddSeq:
return self.run() == 0
# TODO: best interface for "get"
class SwyddGet:
def __call__(
self, proc: str | SwyddProc | SwyddPipe | SwyddSeq, stdout=True, stderr=False
@ -364,11 +359,11 @@ class SwyddGet:
output += result.stderr.strip()
return output
def __lt__(self, proc: str | SwyddProc | SwyddPipe | SwyddSeq) -> str:
return self.__call__(proc)
def __lshift__(self, proc: str | SwyddProc | SwyddPipe | SwyddSeq) -> str:
return self.__call__(proc, stdout=False, stderr=True)
# def __lt__(self, proc: str | SwyddProc | SwyddPipe | SwyddSeq) -> str:
# return self.__call__(proc)
#
# def __lshift__(self, proc: str | SwyddProc | SwyddPipe | SwyddSeq) -> str:
# return self.__call__(proc, stdout=False, stderr=True)
def _get_caller_path() -> Path:
@ -392,22 +387,31 @@ class SwyddSub:
else:
raise ValueError(f"unspported type: {type(exec)}")
def __lt__(self, proc: str | SwyddPipe | SwyddProc | SwyddSeq) -> bool:
return self.__call__(proc)
# TODO: change alias to not confuse with pathlib.Path?
# asset / file ... partial to asset
class SwyddPath:
_root = None
_path = None
def __init__(self, p: Path | None = None) -> None:
if p:
self._path = p
self._path = Path(p)
@classmethod
def __call__(cls, p: str) -> "SwyddPath":
return cls.from_str(p)
@classmethod
def from_str(cls, p: str) -> "SwyddPath":
return cls() / p
def read(self) -> str:
if self._path:
return self._path.read_text()
else:
raise ValueError("path is not set")
def __truediv__(self, p: str | Path) -> "SwyddPath":
if not (root := self._root):
root = _get_caller_path()
@ -427,19 +431,27 @@ class SwyddPath:
raise ValueError("todo")
return self._path
def write(self, txt: str) -> None:
def _write_text(self, txt: str) -> "SwyddPath":
p = self._check()
p.parent.mkdir(exist_ok=True)
p.write_text(txt + "\n")
return self
def append(self, txt: str) -> None:
def write(self, src: "str | SwyddPath") -> "SwyddPath":
if isinstance(src, str):
return self._write_text(src)
elif isinstance(src, SwyddPath):
return self._write_text(src.read())
def _append_text(self, txt: str) -> "SwyddPath":
p = self._check()
p.parent.mkdir(exist_ok=True)
with p.open("a") as f:
f.write(txt)
f.write("\n")
return self
def __mod__(self, dst: "str | SwyddPath | Path") -> None:
def rename(self, dst: "str | SwyddPath | Path") -> None:
if isinstance(dst, str):
dst_p = SwyddPath.from_str(
dst
@ -451,7 +463,7 @@ class SwyddPath:
src_p = self._check()
src_p.rename(dst_p)
def __lt__(self, src: "str | SwyddPath") -> None:
def copy(self, src: "str | SwyddPath") -> None:
if isinstance(src, str):
self.write(src)
elif isinstance(src, SwyddPath):
@ -459,8 +471,11 @@ class SwyddPath:
src_p = src._check()
shutil.copyfile(src_p, dst_p)
def __lshift__(self, txt: str) -> None:
self.append(txt)
def append(self, src: "str | SwyddPath") -> "SwyddPath":
if isinstance(src, str):
return self._append_text(src)
elif isinstance(src, SwyddPath):
return self._append_text(src.read().strip())
def task(func: Callable[..., Any]) -> Callable[..., None]:
@ -573,14 +588,14 @@ def target_generator(
if not (target_path := Path(target)).is_file():
return func(*args, **kwargs)
elif not needs:
sys.stderr.write(f"{target} already exists\n")
sys.stderr.write(f"{target_path} up to date, exiting\n")
else:
target_stats = target_path.stat()
needs_stats = [Path(need).stat() for need in needs]
if any((stat.st_mtime > target_stats.st_mtime for stat in needs_stats)):
return func(*args, **kwargs)
else:
sys.stderr.write("doing nothing\n")
sys.stderr.write(f"{target_path} up to date, exiting\n")
return noop(*args, **kwargs)
@ -736,6 +751,15 @@ def cli() -> None:
SwyddPath(),
)
asset = SwyddPath()
def geterr(*args, **kwargs) -> str:
get_kwargs = dict(stderr=True, stdout=False)
get_kwargs.update(kwargs)
return get(*args, **get_kwargs)
if __name__ == "__main__":
sys.stderr.write("this module should not be invoked directly\n")
sys.exit(1)

21
tasks.py
View file

@ -5,7 +5,7 @@ __import__("sys").path.append("src")
import shutil
import sys
from swydd import cli, get, option, path, sub, task
from swydd import asset, cli, get, option, proc, sub, task
@task
@ -13,35 +13,36 @@ def bootstrap():
"""setup swydd dev environment"""
if not shutil.which("uv"):
sys.exit("uv necessary for swydd development")
sub < "uv sync"
sub < "uv run pre-commit install"
(proc("uv sync").then("uv run pre-commit install").run())
@task
def tests():
"""run pytest"""
sub < "uv run pytest"
sub("uv run pytest")
@task
@option("skip-mypy", "skip mypy")
def check(skip_mypy: bool = False):
"""run pre-commit (and mypy)"""
sub < "uv run pre-commit run --all"
sub("uv run pre-commit run --all")
if not skip_mypy:
sub < "uv run mypy src/"
sub("uv run mypy src/")
@task
def docs():
"""build docs"""
tags = get < "git tag --list"
tags = get("git tag --list")
versions = [line for line in tags.splitlines() if line.startswith("v")]
for tag in versions:
(path / f"docs/{tag}/swydd.py") < (
get < f"git show {tag}:src/swydd/__init__.py"
(
asset(f"docs/{tag}/swydd.py").write(
get(f"git show {tag}:src/swydd/__init__.py")
)
(path / "docs/swydd.py") < (path / "src/swydd/__init__.py")
)
asset("docs/swydd.py").write(asset("src/swydd/__init__.py"))
cli()

44
tests/test_operators.py
View file

@ -1,34 +1,36 @@
from swydd import SwyddPath, get, path, pipe, proc, seq, sub
from swydd import SwyddPath, asset, get, geterr, pipe, proc, seq, sub
def test_operators():
assert sub < "echo 'hello'"
def test_sub():
assert sub("echo 'hello'")
def test_pipes():
assert sub < (
pipe | "cat ../src/swydd/__init__.py" | "grep '__version__'" | "wc -l"
assert sub(
pipe("cat ../src/swydd/__init__.py").pipe("grep '__version__'").pipe("wc -l")
)
assert sub < (
proc | "cat ../src/swydd/__init__.py" | "grep '__version__'" | "wc -l"
assert sub(
proc("cat ../src/swydd/__init__.py").pipe("grep '__version__'").pipe("wc -l")
)
def test_seqs():
# -a is not an arg to cat so the subprocess should return false
assert not sub < (seq & "cat -a" & "echo hello")
assert not sub < (proc & "cat -a" & "echo hello")
assert not sub(seq("cat -a").then("echo hello"))
assert not sub(proc("cat -a").then("echo hello"))
def test_capture():
result = get < "ls src not-src"
result = get("ls src not-src")
assert result == "src:\nswydd"
result = get << "ls src not-src"
result = geterr("ls src not-src")
assert result == "ls: cannot access 'not-src': No such file or directory"
assert "hello part deux" == (get < seq & "echo 'hello'" & "echo hello part deux")
assert "" == (get < (seq & "cp" & "echo hello part deux"))
assert "hello part deux" == get(proc("echo 'hello'").then("echo hello part deux"))
commands = proc("cp").then("echo hello part deux")
assert "" == get(commands)
assert "cp: missing file operand\nTry 'cp --help' for more information." == (
get << (seq & "cp" & "echo hello part deux")
geterr(commands)
)
@ -39,19 +41,19 @@ def check_result_file(file: SwyddPath, text: str) -> bool:
def test_write_to_path():
result_f = path / "products/result.txt"
result_f = asset("products/result.txt")
result_txt = "text to file"
result_f < result_txt
result_f.write(result_txt)
assert check_result_file(result_f, result_txt + "\n")
def test_copy_and_rename():
src_f = path / "fixtures/input.txt"
dst_f = path / "products/input.txt"
dst_f < src_f
src_f = asset("fixtures/input.txt")
dst_f = asset("products/input.txt")
dst_f.copy(src_f)
assert check_result_file(dst_f, "data to copy to another file\n")
dst_f % "products/input2.txt"
dst_f.rename("products/input2.txt")
assert check_result_file(
path / "products/input2.txt", "data to copy to another file\n"
asset("products/input2.txt"), "data to copy to another file\n"
)