Module ds.runner

Run tasks.

Functions

def venv_activate_cmd(venv: pathlib.Path) ‑> str

Return command for activating a .venv

See: https://docs.python.org/3/library/venv.html#how-venvs-work

def find_project(args: Args, task: Task) ‑> Task

Find project-specific dependencies.

Classes

class Runner (args: Args, tasks: Dict[str, ForwardRef('Task')])

Runner(args: ds.args.Args, tasks: Dict[str, ForwardRef('Task')])

Expand source code

@dataclasses.dataclass
class Runner:
    args: Args
    """Command-line arguments."""

    tasks: Tasks
    """Mapping of names to tasks."""

    def run(self, task: Task, override: Task) -> int:
        """Run a `task` overriding parts given `override`."""
        env_from_file = {}
        if task.env_file:
            if not task.env_file.exists():
                log.error(f"Cannot find env-file: {task.env_file}")
                sys.exit(1)

            log.debug(f"Reading env-file: {task.env_file}")
            env_from_file = read_env(task.env_file.read_text())

        resolved = replace(
            override,
            args=task.args + override.args,
            cwd=override.cwd or task.cwd,
            env={**override.env, **task.env},
            _env={**override._env, **override.env, **env_from_file, **task.env},
            env_file=override.env_file or task.env_file,  # for printing
            keep_going=override.keep_going or task.keep_going,
        )

        self.run_pre_post(task, resolved, "pre")

        for dep in task.depends:
            # NOTE: we do not save the return code of any dependencies
            # because they will fail on their own merits.
            self.run(dep, resolved)
        # dependencies ran

        if not task.cmd.strip():  # nothing to do
            return self.run_pre_post(task, resolved, "post")
        # handled dependency-only tasks

        ran, code = self.run_composite(task, resolved)
        if ran:
            return code or self.run_pre_post(task, resolved, "post")
        # composite tasks handled

        # task needs to go into shell
        resolved.cmd = interpolate_args(override.cmd + task.cmd, resolved.args)
        resolved = self.run_in_shell(task, resolved)  # run in shell
        return resolved.code or self.run_pre_post(task, resolved, "post")

    def run_composite(self, task: Task, override: Task) -> Tuple[bool, int]:
        """Run a composite task."""
        ran, code = False, 0
        if not task.name == TASK_COMPOSITE:
            return ran, code

        cmd, *args = split(task.cmd)
        others = glob_names(self.tasks.keys(), cmd.split(GLOB_DELIMITER))
        for name in others:
            other = self.tasks.get(name)
            # - name is of another task
            # - task is not trying to run itself (ls: ['ls'] runs in shell)
            # - task is not in the other's dependencies
            if other and other != task and task not in other.depends:
                ran = True
                code = self.run(other, replace(override, args=override.args + args))
            # in all other cases, we're going to run this in the shell
        return ran, code

    def run_pre_post(self, task: Task, override: Task, pre_post: str = "pre") -> int:
        """Run pre- or post- task."""
        name = task.name
        if not name or name == TASK_COMPOSITE or not getattr(self.args, pre_post):
            return 0

        for check in [f"{pre_post}{name}", f"{pre_post}_{name}", f"{pre_post}-{name}"]:
            log.debug(f"check {check}")
            if sub_task := self.tasks.get(check):
                log.debug(f"EXPERIMENTAL: Running --{pre_post} task {check}")
                return self.run(sub_task, override)

        # no task found
        return 0

    def run_in_shell(self, task: Task, resolved: Task) -> Task:
        """Run the resolved task in the shell."""
        dry_run = self.args.dry_run
        task.pprint(resolved, dry_run)
        if dry_run:  # do not actually run the command
            return resolved

        combined_env = {**ENV, **resolved.env, **resolved._env}
        proc = subprocess.run(
            resolved.cmd,
            shell=True,
            text=True,
            cwd=resolved.cwd,
            env=combined_env,
            executable=combined_env.get("SHELL", combined_env.get("COMSPEC")),
        )

        resolved.code = proc.returncode
        if resolved.code != 0 and not resolved.keep_going:
            log.error(f"return code = {resolved.code}")
            sys.exit(resolved.code)
        return resolved

Class variables

var args : Args

Command-line arguments.

var tasks : Dict[str, Task]

Mapping of names to tasks.

Methods

def run(self, task: Task, override: Task) ‑> int

Run a task overriding parts given override.

def run_composite(self, task: Task, override: Task) ‑> Tuple[bool, int]

Run a composite task.

def run_pre_post(self, task: Task, override: Task, pre_post: str = 'pre') ‑> int

Run pre- or post- task.

def run_in_shell(self, task: Task, resolved: Task) ‑> Task

Run the resolved task in the shell.