This commit is contained in:
Daylin Morgan 2024-07-09 12:17:21 -05:00
parent 3c6afdd5cb
commit a7ac1dd094
Signed by: daylin
GPG key ID: 950D13E9719334AD

View file

@ -90,6 +90,7 @@ class Graph:
class Context: class Context:
def __init__(self) -> None: def __init__(self) -> None:
self._env: Dict[str, str] = {}
self._tasks: Dict[str, Any] = {} self._tasks: Dict[str, Any] = {}
self.targets: Dict[str, Any] = {} self.targets: Dict[str, Any] = {}
self.data: Any = None self.data: Any = None
@ -147,6 +148,10 @@ class Context:
ctx = Context() ctx = Context()
def define_env(key: str, value: str) -> None:
ctx._env.update({key: value})
class Exec: class Exec:
def __init__(self, cmd: str, shell: bool = False) -> None: def __init__(self, cmd: str, shell: bool = False) -> None:
self.shell = shell self.shell = shell
@ -156,9 +161,13 @@ class Exec:
if ctx.verbose: if ctx.verbose:
sys.stdout.write(f"swydd exec | {self.cmd}\n") sys.stdout.write(f"swydd exec | {self.cmd}\n")
if self.shell: if self.shell:
return subprocess.run(self.cmd, shell=True).returncode return subprocess.run(
self.cmd, shell=True, env={**os.environ, **ctx._env}
).returncode
else: else:
return subprocess.run(shlex.split(self.cmd)).returncode return subprocess.run(
shlex.split(self.cmd), env={**os.environ, **ctx._env}
).returncode
def sh(cmd: str, shell: bool = False) -> int: def sh(cmd: str, shell: bool = False) -> int:
@ -174,21 +183,38 @@ def task(func: Callable[..., Any]) -> Callable[..., None]:
return wrap return wrap
# def inspect_wrapper(place, func): def inspect_wrapper(place, func):
# if wrapped := getattr(func, "__wrapped__", None): if wrapped := getattr(func, "__wrapped__", None):
# print(place, "wrapped->", id(wrapped)) print(place, "wrapped->", id(wrapped))
#
# print( print(
# place, place,
# id(func), id(func),
# ) )
#
def task2(
hidden: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Callable[..., None]]]:
def wrapper(func: Callable[..., Any]) -> Callable[..., Callable[..., None]]:
ctx._add_task(func, show=True)
inspect_wrapper("task", func)
@wraps(func)
def inner(*args: Any, **kwargs: Any) -> Callable[..., None]:
return func(*args, **kwargs)
return inner
return wrapper
def targets( def targets(
*args: str, *args: str,
) -> Callable[[Callable[..., Any]], Callable[..., Callable[..., None]]]: ) -> Callable[[Callable[..., Any]], Callable[..., Callable[..., None]]]:
def wrapper(func: Callable[..., Any]) -> Callable[..., Callable[..., None]]: def wrapper(func: Callable[..., Any]) -> Callable[..., Callable[..., None]]:
inspect_wrapper("targets", func)
ctx._add_task(func) ctx._add_task(func)
for arg in args: for arg in args:
ctx._add_target(func, arg) ctx._add_target(func, arg)
@ -300,7 +326,7 @@ def generate_task_subparser(
args = (f"--{name.replace('_','-')}",) args = (f"--{name.replace('_','-')}",)
kwargs = {"help": info.get("help", "")} kwargs = {"help": info.get("help", "")}
if param.annotation == bool: if param.annotation is bool:
kwargs.update({"default": False, "action": "store_true"}) kwargs.update({"default": False, "action": "store_true"})
elif param.annotation != Parameter.empty: elif param.annotation != Parameter.empty:
kwargs.update({"type": param.annotation}) kwargs.update({"type": param.annotation})