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!

Browse source
This commit is contained in:
Daylin Morgan 2024-10-24 16:44:08 -05:00
parent 83f94cd4c2
commit 2594234e76
Signed by: daylin
GPG key ID: 950D13E9719334AD
1 changed files with 188 additions and 159 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

347
src/swydd/__init__.py
View file

@ -162,7 +162,7 @@ class Context:
return self._tasks[self._ids[name]] return self._tasks[self._ids[name]]
class SwyddFlags: class Flags:
def __init__(self) -> None: def __init__(self) -> None:
self._flags: Dict[str, Any] = {} self._flags: Dict[str, Any] = {}
self._flag_defs: List[Tuple[Tuple[str, ...], Any]] = [] self._flag_defs: List[Tuple[Tuple[str, ...], Any]] = []
@ -189,10 +189,10 @@ class SwyddFlags:
ctx = Context() ctx = Context()
flags = SwyddFlags() flags = Flags()
class SwyddSubResult: class SubResult:
def __init__( def __init__(
self, self,
code: int, code: int,
@ -206,7 +206,7 @@ class SwyddSubResult:
self._proces = process self._proces = process
@classmethod @classmethod
def from_completed_process(cls, p: CompletedProcess) -> "SwyddSubResult": def from_completed_process(cls, p: CompletedProcess) -> "SubResult":
return cls(p.returncode, p.stdout, p.stderr, p) return cls(p.returncode, p.stdout, p.stderr, p)
@classmethod @classmethod
@ -215,49 +215,60 @@ class SwyddSubResult:
p: Popen, p: Popen,
stdout: str = "", stdout: str = "",
stderr: str = "", stderr: str = "",
) -> "SwyddSubResult": ) -> "SubResult":
return cls(p.returncode, stdout, stderr, p) return cls(p.returncode, stdout, stderr, p)
class SwyddProc: class Proc:
def __init__( """a class"""
self, cmd: str | None = None, output: bool = False, **kwargs: Any
) -> None: def __init__(self, cmd: str, **kwargs: Any) -> None:
self._cmd = cmd self._cmd = cmd
if cmd: if cmd:
self.cmd = shlex.split(cmd) self.cmd = shlex.split(cmd)
self.output = output
self.cmd_kwargs = kwargs self.cmd_kwargs = kwargs
@classmethod def pipe(self, proc: str | Proc) -> ProcPipe:
def __call__(cls, *args, **kwargs) -> "SwyddProc": """ProcPipe the output into `proc`
return cls(*args, **kwargs)
Parameters
----------
proc
Command to feed stdout into
Returns
-------
ProcPipe
"""
def pipe(self, proc: "str | SwyddProc") -> "SwyddPipe | SwyddProc":
if isinstance(proc, str): if isinstance(proc, str):
if self._cmd is None: return ProcPipe(self, proc)
return SwyddPipe(SwyddProc(proc)) elif isinstance(proc, Proc):
else: return ProcPipe(proc)
return SwyddPipe(self, proc)
elif isinstance(proc, SwyddProc):
return SwyddPipe(proc)
def then(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq": def then(self, proc: str | Proc | ProcSeq) -> ProcSeq:
if self._cmd: """If successful execute next `proc`
return SwyddSeq(self, proc)
if isinstance(proc, SwyddProc): Parameters
return SwyddSeq(proc) ----------
proc
Command to execute after
Returns
-------
ProcSeq
"""
if isinstance(proc, Proc):
return ProcSeq(proc)
# should swydd seq even be supported here? # should swydd seq even be supported here?
elif isinstance(proc, SwyddSeq): elif isinstance(proc, ProcSeq):
return proc return proc
else: else:
return SwyddSeq(SwyddProc(proc)) return ProcSeq(Proc(proc))
def _build_kwargs(self) -> Dict[str, Any]: def _build_kwargs(self, output: bool) -> Dict[str, Any]:
sub_kwargs: Dict[str, Any] = dict(env={**os.environ, **ctx._env}) sub_kwargs: Dict[str, Any] = dict(env={**os.environ, **ctx._env})
if self.output: if output:
sub_kwargs["text"] = True # assume text is the desired output sub_kwargs["text"] = True # assume text is the desired output
sub_kwargs["stdout"] = PIPE sub_kwargs["stdout"] = PIPE
@ -268,11 +279,10 @@ class SwyddProc:
if ctx.verbose: if ctx.verbose:
sys.stdout.write(f"swydd exec | {self._cmd}\n") sys.stdout.write(f"swydd exec | {self._cmd}\n")
def execute(self, output: bool = False) -> SwyddSubResult: def execute(self, output: bool = False) -> SubResult:
self.output = self.output or output
self._show_command() self._show_command()
p = Popen(self.cmd, **self._build_kwargs()) p = Popen(self.cmd, **self._build_kwargs(output))
try: try:
out, err = p.communicate() out, err = p.communicate()
@ -283,27 +293,27 @@ class SwyddProc:
p.wait() p.wait()
out, err = p.communicate() out, err = p.communicate()
return SwyddSubResult.from_popen(p, out, err) return SubResult.from_popen(p, out, err)
def check(self) -> bool: def check(self) -> bool:
return self.execute().code == 0 return self.execute().code == 0
class SwyddPipe: class ProcPipe:
def __init__(self, *procs: "str | SwyddProc | SwyddPipe") -> None: """
prefer: :func:`Proc.pipe`
"""
def __init__(self, *procs: str | Proc | ProcPipe) -> None:
self._procs = [] self._procs = []
for proc in procs: for proc in procs:
if isinstance(proc, str): if isinstance(proc, str):
self._procs.append(SwyddProc(proc)) self._procs.append(Proc(proc))
elif isinstance(proc, SwyddProc): elif isinstance(proc, Proc):
self._procs.append(proc) self._procs.append(proc)
elif isinstance(proc, SwyddPipe): elif isinstance(proc, ProcPipe):
self._procs.extend(proc._procs) self._procs.extend(proc._procs)
@classmethod
def __call__(cls, *args, **kwargs) -> "SwyddPipe":
return cls(*args, **kwargs)
def check(self) -> bool: def check(self) -> bool:
return self.execute().code == 0 return self.execute().code == 0
@ -312,7 +322,7 @@ class SwyddPipe:
cmd_str = " | ".join([p._cmd for p in self._procs]) cmd_str = " | ".join([p._cmd for p in self._procs])
sys.stdout.write(f"swydd exec | {cmd_str}\n") sys.stdout.write(f"swydd exec | {cmd_str}\n")
def execute(self, output: bool = False) -> SwyddSubResult: def execute(self, output: bool = False) -> SubResult:
procs = [] procs = []
sub_kwargs: Dict[str, Any] = dict(env={**os.environ, **ctx._env}) sub_kwargs: Dict[str, Any] = dict(env={**os.environ, **ctx._env})
self._show_command() self._show_command()
@ -343,36 +353,36 @@ class SwyddPipe:
procs[-1].wait() procs[-1].wait()
out, err = procs[-1].communicate() out, err = procs[-1].communicate()
return SwyddSubResult.from_popen(procs[-1], out, err) return SubResult.from_popen(procs[-1], out, err)
def pipe(self, proc: "str | SwyddProc | SwyddPipe") -> "SwyddPipe": def pipe(self, proc: "str | Proc | ProcPipe") -> "ProcPipe":
return SwyddPipe(self, proc) return ProcPipe(self, proc)
class SwyddSeq: class ProcSeq:
def __init__(self, *procs: "str | SwyddProc | SwyddSeq") -> None: """
prefer: :func:`Proc.then`
"""
def __init__(self, *procs: "str | Proc | ProcSeq") -> None:
self._procs = [] self._procs = []
for proc in procs: for proc in procs:
if isinstance(proc, SwyddSeq): if isinstance(proc, ProcSeq):
self._procs.extend(self._procs) self._procs.extend(self._procs)
if isinstance(proc, SwyddProc): if isinstance(proc, Proc):
self._procs.append(proc) self._procs.append(proc)
elif isinstance(proc, str): elif isinstance(proc, str):
self._procs.append(SwyddProc(proc)) self._procs.append(Proc(proc))
def _show_command(self) -> None: def _show_command(self) -> None:
if ctx.verbose: if ctx.verbose:
cmd_str = " && ".join([p._cmd for p in self._procs]) cmd_str = " && ".join([p._cmd for p in self._procs])
sys.stderr.write(f"sywdd exec | {cmd_str}\n") sys.stderr.write(f"sywdd exec | {cmd_str}\n")
@classmethod def then(self, proc: "str | Proc | ProcSeq") -> "ProcSeq":
def __call__(cls, *args, **kwargs) -> "SwyddSeq": return ProcSeq(*self._procs, proc)
return cls(*args, **kwargs)
def then(self, proc: "str | SwyddProc | SwyddSeq") -> "SwyddSeq": def execute(self, output: bool = False) -> "SubResult":
return SwyddSeq(*self._procs, proc)
def execute(self, output: bool = False) -> "SwyddSubResult":
self._show_command() self._show_command()
results = [] results = []
@ -396,30 +406,6 @@ class SwyddSeq:
return self.run() == 0 return self.run() == 0
# TODO: best interface for "get"
class SwyddGet:
def __call__(
self, proc: str | SwyddProc | SwyddPipe | SwyddSeq, stdout=True, stderr=False
) -> str:
if isinstance(proc, str):
result = SwyddProc(proc, output=True).execute()
elif isinstance(proc, SwyddPipe):
result = proc.execute(output=True)
elif isinstance(proc, SwyddProc):
result = proc.execute()
elif isinstance(proc, SwyddSeq):
result = proc.execute(output=True)
else:
raise NotImplementedError(f"not implemented for type: {type(exec)}")
output = ""
if stdout and result.stdout:
output += result.stdout.strip()
if stderr and result.stderr:
output += result.stderr.strip()
return output
def _get_caller_path() -> Path: def _get_caller_path() -> Path:
# NOTE: jupyter will hate this code I'm sure # NOTE: jupyter will hate this code I'm sure
for i, frame in enumerate(inspect.stack()): for i, frame in enumerate(inspect.stack()):
@ -428,35 +414,14 @@ def _get_caller_path() -> Path:
raise ValueError("failed to find root directory of runner") raise ValueError("failed to find root directory of runner")
class SwyddSub: # Asset(ag) / f
def __call__(self, proc: str | SwyddPipe | SwyddProc | SwyddSeq) -> bool: # Asset / a <- also support!
if isinstance(proc, str):
return SwyddProc(proc).check()
elif (
isinstance(proc, SwyddProc)
or isinstance(proc, SwyddSeq)
or isinstance(proc, SwyddPipe)
):
return proc.check()
else:
raise ValueError(f"unspported type: {type(exec)}")
class SwyddPath: class Asset:
_root = None def __init__(self, p: str) -> None:
_path = None self._root = _get_caller_path()
self._path = self._root / p
def __init__(self, p: Path | None = None) -> None:
if 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: def read(self) -> str:
if self._path: if self._path:
@ -464,38 +429,28 @@ class SwyddPath:
else: else:
raise ValueError("path is not set") raise ValueError("path is not set")
def __truediv__(self, p: str | Path) -> "SwyddPath": def __truediv__(self, p: str) -> "Asset":
if not (root := self._root): self._path = self._path / p
root = _get_caller_path() return self
if not self._path:
if isinstance(p, str):
return SwyddPath(root / p)
elif isinstance(p, Path):
return SwyddPath(p)
else:
og = self._path.relative_to(root)
return SwyddPath(og / p)
def _check(self) -> Path: def _check(self) -> Path:
if self._path is None: if self._path is None:
raise ValueError("todo") raise ValueError("todo")
return self._path return self._path
def _write_text(self, txt: str) -> "SwyddPath": def _write_text(self, txt: str) -> "Asset":
p = self._check() p = self._check()
p.parent.mkdir(exist_ok=True) p.parent.mkdir(exist_ok=True)
p.write_text(txt + "\n") p.write_text(txt + "\n")
return self return self
def write(self, src: "str | SwyddPath") -> "SwyddPath": def write(self, src: "str | Asset") -> "Asset":
if isinstance(src, str): if isinstance(src, str):
return self._write_text(src) return self._write_text(src)
elif isinstance(src, SwyddPath): elif isinstance(src, Asset):
return self._write_text(src.read()) return self._write_text(src.read())
def _append_text(self, txt: str) -> "SwyddPath": def _append_text(self, txt: str) -> "Asset":
p = self._check() p = self._check()
p.parent.mkdir(exist_ok=True) p.parent.mkdir(exist_ok=True)
with p.open("a") as f: with p.open("a") as f:
@ -503,11 +458,9 @@ class SwyddPath:
f.write("\n") f.write("\n")
return self return self
def rename(self, dst: "str | SwyddPath | Path") -> None: def rename(self, dst: "str | Asset | Path") -> None:
if isinstance(dst, str): if isinstance(dst, str):
dst_p = SwyddPath.from_str( dst_p = Asset(dst)._check()
dst
)._check() # <- TODO: ensure this uses self._root?
elif isinstance(dst, Path): elif isinstance(dst, Path):
dst_p = dst dst_p = dst
else: else:
@ -515,18 +468,18 @@ class SwyddPath:
src_p = self._check() src_p = self._check()
src_p.rename(dst_p) src_p.rename(dst_p)
def copy(self, src: "str | SwyddPath") -> None: def copy(self, src: "str | Asset") -> None:
if isinstance(src, str): if isinstance(src, str):
self.write(src) self.write(src)
elif isinstance(src, SwyddPath): elif isinstance(src, Asset):
dst_p = self._check() dst_p = self._check()
src_p = src._check() src_p = src._check()
shutil.copyfile(src_p, dst_p) shutil.copyfile(src_p, dst_p)
def append(self, src: "str | SwyddPath") -> "SwyddPath": def append(self, src: "str | Asset") -> "Asset":
if isinstance(src, str): if isinstance(src, str):
return self._append_text(src) return self._append_text(src)
elif isinstance(src, SwyddPath): elif isinstance(src, Asset):
return self._append_text(src.read().strip()) return self._append_text(src.read().strip())
@ -543,6 +496,8 @@ def _inspect_wrapper(place, func):
def task( def task(
arg: Callable[..., Any] | None = None, arg: Callable[..., Any] | None = None,
): ):
"""decorator to convert a function into a swydd task"""
def wrapper( def wrapper(
func: Callable[..., Any] | None = None, func: Callable[..., Any] | None = None,
) -> Callable[..., Callable[..., None]]: ) -> Callable[..., Callable[..., None]]:
@ -604,6 +559,18 @@ def option(
short: str = "", short: str = "",
**help_kwargs: str, **help_kwargs: str,
) -> Callable[[Callable[..., Any]], Callable[..., Callable[..., None]]]: ) -> Callable[[Callable[..., Any]], Callable[..., Callable[..., None]]]:
"""add help and additional args for option
Parameters
----------
name
variable name used in wrapped function, subsitute underscores with hypens
help
help description for variable
**help_kwargs
kwargs which will be passed onto :func:`argparse.ArgumentParser.add_argument`
"""
def wrapper(func: Callable[..., Any]) -> Callable[..., Callable[..., None]]: def wrapper(func: Callable[..., Any]) -> Callable[..., Callable[..., None]]:
ctx._update_option(func, name.replace("-", "_"), help, short, **help_kwargs) ctx._update_option(func, name.replace("-", "_"), help, short, **help_kwargs)
@ -726,7 +693,7 @@ def _generate_task_subparser(
def executor(*args, **kwargs): def executor(*args, **kwargs):
for need in task.needs: for need in task.needs:
asset(need)._check() Asset(need)._check()
f = ( f = (
_target_generator(target, ctx._graph.nodes[target])(task.func) _target_generator(target, ctx._graph.nodes[target])(task.func)
@ -771,6 +738,13 @@ def _add_targets(
def cli(default: str | None = None) -> None: def cli(default: str | None = None) -> None:
"""activate swydd cli
Parameters
----------
default
The default args passed line passed on to sywdd.
"""
ctx._generate_graph() ctx._generate_graph()
if len(sys.argv) > 1 and sys.argv[1] == "+swydd": if len(sys.argv) > 1 and sys.argv[1] == "+swydd":
_internal_cli() _internal_cli()
@ -834,48 +808,103 @@ def cli(default: str | None = None) -> None:
f(**args) f(**args)
( def get(proc: str | Proc | ProcPipe | ProcSeq, stdout=True, stderr=False) -> str:
proc, """execute subprocess and capture outputs
pipe,
seq, see also: :func:`geterr`
sub,
get, Parameters
asset, ----------
) = ( proc
SwyddProc(), Command to execute.
SwyddPipe(), stdout
SwyddSeq(), If true, capture stdout
SwyddSub(), stderr
SwyddGet(), If true, capture stderr
SwyddPath(),
) Returns
-------
str
Captured output
"""
if isinstance(proc, str):
result = Proc(proc).execute(output=True)
elif isinstance(proc, ProcPipe):
result = proc.execute(output=True)
elif isinstance(proc, Proc):
result = proc.execute()
elif isinstance(proc, ProcSeq):
result = proc.execute(output=True)
else:
raise NotImplementedError(f"not implemented for type: {type(exec)}")
output = ""
if stdout and result.stdout:
output += result.stdout.strip()
if stderr and result.stderr:
output += result.stderr.strip()
return output
def geterr(*args, **kwargs) -> str: def geterr(
get_kwargs = dict(stderr=True, stdout=False) proc: str | Proc | ProcPipe | ProcSeq,
get_kwargs.update(kwargs) ) -> str:
return get(*args, **get_kwargs) """execute subprocess and return stderr
Parameters
----------
proc : str, :class:`.Proc`, :class:`.ProcSeq`, :class:`.ProcPipe`
Command to execute.
Returns
-------
str
Output captured from stderr.
"""
return get(proc, stdout=True, stderr=False)
# TODO: think more about how "rest" should work...
def sub(proc: str | Proc | ProcPipe | ProcSeq, rest: bool = False) -> bool:
if isinstance(proc, str):
cmd = proc if not rest else (proc + " " + " ".join(ctx.rest))
return Proc(cmd).check()
elif (
isinstance(proc, Proc)
or isinstance(proc, ProcSeq)
or isinstance(proc, ProcPipe)
):
if rest:
raise ValueError("rest is only supported when passing str")
return proc.check()
else:
raise ValueError(f"unspported type: {type(exec)}")
def setenv(key: str, value: str) -> None: def setenv(key: str, value: str) -> None:
"""Set environment variable shared by all Procs"""
ctx._env.update({key: value}) ctx._env.update({key: value})
__all__ = [ __all__ = [
"proc",
"pipe",
"seq",
"sub", "sub",
"get", "get",
"asset",
"ctx", "ctx",
"geterr", "geterr",
"setenv", "setenv",
"cli", "cli",
"task", "task",
"targets",
"needs",
"option",
"flags", "flags",
"Proc",
"ProcPipe",
"ProcSeq",
"Asset",
] ]
if __name__ == "__main__": if __name__ == "__main__":
sys.stderr.write("this module should not be invoked directly\n") sys.stderr.write("this module should not be invoked directly\n")
sys.exit(1) sys.exit()