Module ds.tasks

Parse and run tasks.

Global variables

var Tasks

Mapping of task names to Task objects.

var ORIGINAL_CWD

Save a reference to the original working directory.

Functions

def check_cycles(tasks: Tasks) ‑> List[str]

Raise a CycleError if there is a cycle in the task graph.

def print_tasks(path: Path, tasks: Tasks) ‑> None

Pretty print task names.

Classes

class CycleError (*args, **kwargs)

Subclass of ValueError raised by TopologicalSorter.prepare if cycles exist in the working graph.

If multiple cycles exist, only one undefined choice among them will be reported and included in the exception. The detected cycle can be accessed via the second element in the args attribute of the exception instance and consists in a list of nodes, such that each node is, in the graph, an immediate predecessor of the next node in the list. In the reported list, the first and the last node will be the same, to make it clear that it is cyclic.

Expand source code

class CycleError(ValueError):
    """Subclass of ValueError raised by TopologicalSorter.prepare if cycles
    exist in the working graph.

    If multiple cycles exist, only one undefined choice among them will be reported
    and included in the exception. The detected cycle can be accessed via the second
    element in the *args* attribute of the exception instance and consists in a list
    of nodes, such that each node is, in the graph, an immediate predecessor of the
    next node in the list. In the reported list, the first and the last node will be
    the same, to make it clear that it is cyclic.
    """

    pass

Ancestors

• builtins.ValueError
• builtins.Exception
• builtins.BaseException

class Task (origin: Optional[Path] = None, origin_key: str = '', name: str = '', help: str = '', verbatim: bool = False, depends: List[Task] = <factory>, cmd: str = '', code: int = 0, args: List[str] = <factory>, cwd: Optional[Path] = None, env: Dict[str, str] = <factory>, env_file: Optional[Path] = None, keep_going: bool = False)

Represents a thing to be done.

Expand source code

@dataclass
class Task:
    """Represents a thing to be done."""

    origin: Optional[Path] = None
    """File from which this configuration came."""

    origin_key: str = ""
    """Key from which this task came."""

    name: str = ""
    """Task name."""

    help: str = ""
    """Task description."""

    verbatim: bool = False
    """Whether to format the command at all."""

    depends: List[Task] = field(default_factory=list)
    """Tasks to execute before this one."""

    cmd: str = ""
    """Shell command to execute after `depends`."""

    code: int = 0
    """Return code from running this task."""

    # NOTE: args, cwd, env, keep_going are overridable
    # via the CLI or when calling a composite command.

    args: List[str] = field(default_factory=list)
    """Additional arguments to `cmd`."""

    cwd: Optional[Path] = None
    """Task working directory."""

    env: Dict[str, str] = field(default_factory=dict)
    """Task environment variables."""

    _env: Dict[str, str] = field(default_factory=dict)
    """Hidden environment variables."""

    env_file: Optional[Path] = None
    """Path to an environment file to load."""

    keep_going: bool = False
    """Ignore a non-zero return code."""

    def pprint(self, override: Optional[Task] = None, dry_run: bool = False) -> None:
        """Print a representation of this task."""
        is_run = override or dry_run
        display = self
        if override:
            display = replace(self, cmd=override.cmd, keep_going=override.keep_going)

        print()

        if dry_run:
            print("[DRY RUN]")
        if display.help:
            print("#", display.help)
        print(">", wrap_cmd(self.as_args(override)))

        if not is_run and display.depends:
            print(
                [
                    f"{TASK_KEEP_GOING if t.keep_going else ''}{t.cmd}"
                    for t in display.depends
                ]
            )

        if display.cmd:
            if display.verbatim:
                print("$", display.cmd.strip().replace("\n", "\n$ "))
            else:
                print(f"$ {wrap_cmd(display.cmd)}")

    def as_args(self, override: Optional[Task] = None) -> str:
        """Return a shell representation of running this task."""
        override = override or Task()

        args = ["ds"]
        if self.cwd:
            args.extend(["--cwd", str(self.cwd)])
        if self.env_file:
            args.extend(["--env-file", str(self.env_file)])
        for key, val in (self.env or {}).items():
            args.extend(["-e", f"{key}={val}"])

        prefix = ""
        if self.keep_going or override.keep_going:
            prefix = TASK_KEEP_GOING
        if self.name == TASK_COMPOSITE:
            args.append(f"{prefix}{self.cmd}")
        elif self.name:
            args.append(f"{prefix}{self.name}")
        return join(args)

Class variables

var depends : List[Task]

Tasks to execute before this one.

var args : List[str]

Additional arguments to cmd.

var env : Dict[str, str]

Task environment variables.

var origin : Optional[pathlib.Path]

File from which this configuration came.

var origin_key : str

Key from which this task came.

var name : str

Task name.

var help : str

Task description.

var verbatim : bool

Whether to format the command at all.

var cmd : str

Shell command to execute after depends.

var code : int

Return code from running this task.

var cwd : Optional[pathlib.Path]

Task working directory.

var env_file : Optional[pathlib.Path]

Path to an environment file to load.

var keep_going : bool

Ignore a non-zero return code.

Methods

def pprint(self, override: Optional[Task] = None, dry_run: bool = False) ‑> None

Print a representation of this task.

def as_args(self, override: Optional[Task] = None) ‑> str

Return a shell representation of running this task.