Running from a CLI

The tools of GAIn run their task graphs from a command line through TaskGraphCli. A script of your own can do the same. TaskGraphCli adds the task graph options to an argparse parser, and it runs a task graph by the parsed arguments. This page describes the Python side of the options. The user page Running in parallel describes what each option does.

Creating the arguments

TaskGraphCli.add_arguments adds the options to a parser:

import argparse

from gain.task_graph import TaskGraphCli

parser = argparse.ArgumentParser()
TaskGraphCli.add_arguments(parser)
args = parser.parse_args()

Each option adds a key to the parsed arguments. Several keys do not have the name of their option: --process-pool adds use_process_pool, -N adds dask_cluster_name, -c adds dask_cluster_config_file and -d adds task_status_dir. The docstring of add_arguments lists every key. add_arguments adds no -v. The commands list and status read verbose, so add -v with VerbosityConfiguration.set_arguments.

Three keywords of add_arguments change the options:

use_commands=False

Removes the optional command run, list or status. The script then always runs the graph.

task_progress_mode=False

Removes --force and -d. The parsed arguments then have no force and no task_status_dir.

Warning

Pass the same task_progress_mode to add_arguments and to process_graph. process_graph has its own keyword. If you pass task_progress_mode=False to add_arguments only, process_graph still builds a FileTaskCache, and the task cache writes its records into the current directory.

default_task_status_dir

Sets the default of -d, the task status directory. The default is ./.task-progress. With None, a run without -d writes the records of the task cache into the current directory.

The lifecycle

TaskGraphCli.process_graph takes the task graph and the parsed arguments as keywords. It does these steps:

1. It prunes the graph to the task ids of -t, and to the tasks that they depend on.

2. It builds the task cache. With task_progress_mode true, the default, the task cache is a FileTaskCache over the task status directory.

3. For the command run, the default, it creates the task executor with create_executor(). It runs the graph inside the context of the task executor, and it returns true when every task succeeds.

4. For the commands list and status, it prints the record type of each task and runs nothing. It returns true.

--force does not change the task cache. The task executor then runs every task and writes every record again.

create_executor picks the task executor by the parsed arguments:

• -j 1 gives a SequentialExecutor.

• --process-pool with -j N gives a ProcessPoolTaskExecutor with N processes. Without -j, the pool has one process for each CPU.

• Every other case gives a DaskExecutor on a named cluster, also a command line without -j.

-j 1 together with -N or -c fails an assert in create_executor. A direct call of create_executor() without keywords gives a SequentialExecutor.

A running example, part 3

Part 1, on the page Graphs, builds a graph that counts the N bases of a reference genome. Part 2, on the page Task executors, runs it from Python. Part 3 runs it from a command line.

The script has a main(argv) function. It adds the genomic context options with context_providers_add_argparser_arguments and the task graph options with add_arguments. Then it parses argv and builds the genomic context. It takes the resource of the reference genome from the genomic context, builds the graph of part 1, and calls process_graph:

def main(argv: list[str] | None = None) -> bool:
    """Count the ``N`` bases of a reference genome from a command line."""
    parser = argparse.ArgumentParser(
        description="Count the N bases of a reference genome.")
    context_providers_add_argparser_arguments(parser)
    TaskGraphCli.add_arguments(parser)
    args = parser.parse_args(argv)

    context = build_cli_genomic_context(vars(args))
    genome = context.get_reference_genome()
    if genome is None:
        parser.error("name the reference genome with -R")

    graph, _ = build_n_count_graph(genome.resource)
    return TaskGraphCli.process_graph(graph, **vars(args))

A command line names the GRR with --grr-directory <dir> or -g <grr file>, and the reference genome with -R <genome resource id>.

The return value of main is not an exit status: main returns True when every task succeeds. A console script that calls sys.exit(main()) exits with status 1 after a successful run. Map the return value to an exit status in the function that the console script calls:

def count_n() -> None:
    """Run ``main`` and exit with status 0 only when every task succeeds."""
    sys.exit(0 if main(sys.argv[1:]) else 1)

In the commands below, count_n is a console script that calls the function count_n. The first command runs the graph sequentially and writes the records into the task status directory task-status:

count_n --grr-directory grr -R genome -j 1 -d task-status

After the run, the command status prints COMPUTED for each task:

count_n --grr-directory grr -R genome status -j 1 -d task-status

API

TaskGraphCli is defined in the module gain.task_graph.cli_tools. It is also exported by the package gain.task_graph, as gain.task_graph.TaskGraphCli.

TaskGraphCli()	Add the task graph options to a command line and run a task graph.
TaskGraphCli.add_arguments(parser, *[, ...])	Add the task graph options to a command line parser.
TaskGraphCli.create_executor([task_cache])	Create the task executor that the parsed arguments choose.
TaskGraphCli.process_graph(task_graph, *[, ...])	Run a task graph, or report its tasks, by the parsed arguments.

class TaskGraphCli

Add the task graph options to a command line and run a task graph.

A tool calls add_arguments() on its parser, and then calls process_graph() with its task graph and the parsed arguments as keywords. The page Running in parallel describes the options for the user.

static add_arguments(parser: ArgumentParser, *, task_progress_mode: bool = True, default_task_status_dir: str | None = './.task-progress', use_commands: bool = True) → None

Add the task graph options to a command line parser.

The options choose the task executor, the number of workers, the task status directory and the command. Pass the parsed arguments to process_graph() with **vars(args). Several keys do not have the name of their option. The options add these keys:

• -j adds jobs.

• --process-pool adds use_process_pool.

• -N adds dask_cluster_name.

• -c adds dask_cluster_config_file.

• --task-log-dir adds task_log_dir.

• The optional command adds command.

• -t adds task_ids.

• --keep-going adds keep_going.

• --fork-tasks adds fork_tasks.

• --force adds force.

• -d adds task_status_dir.

The function adds no -v. The commands list and status read verbose, so add it with set_arguments().

Parameters:

• parser (argparse.ArgumentParser) – The parser that gets the options.

• task_progress_mode (bool) – When false, the parser gets no --force and no -d, and the parsed arguments have no force and no task_status_dir. Pass the same value to process_graph(): it builds a FileTaskCache by default whatever the parser holds, and without task_status_dir the task cache writes its record files into the current directory.

• default_task_status_dir (str | None) – The default of -d, the task status directory. None means that a run without -d writes the record files of the task cache into the current directory.

• use_commands (bool) – When false, the parser gets no optional command run, list or status, and process_graph() always runs the graph.

static create_executor(task_cache: TaskCache | None = None, **kwargs: Any) → TaskGraphExecutor

Create the task executor that the parsed arguments choose.

With jobs equal to 1, the function creates a SequentialExecutor. Otherwise, with use_process_pool true, it creates a ProcessPoolTaskExecutor with jobs processes. In every other case, it creates a DaskExecutor on a named cluster.

Parameters:

• task_cache (TaskCache | None) – The task cache of the task executor. None means a NoTaskCache, which runs every task and records nothing.

• **kwargs –

  The parsed arguments of add_arguments(). The function reads these keys, and it passes all of them to the task executor too:

  jobs

  The number of workers. A missing key means 1, so a call without keywords creates a SequentialExecutor. A parsed command line always holds jobs, with the value None when -j is absent, so a command line without -j creates a DaskExecutor, or a ProcessPoolTaskExecutor with --process-pool. None means one process for each CPU for the process pool, and the default of the named cluster for dask.

  use_process_pool

  When true, and jobs is not 1, the function creates a ProcessPoolTaskExecutor.

  dask_cluster_name

  The named cluster of the DaskExecutor. None means the default named cluster.

  dask_cluster_config_file

  A cluster config file that describes the cluster of the DaskExecutor. None means that the function uses dask_cluster_name.

Returns:

The new task executor. Use it as a context manager, so that it releases its workers.

Return type:

TaskGraphExecutor

Raises:

AssertionError – When jobs is 1, or use_process_pool is true, and dask_cluster_name or dask_cluster_config_file is set. Also when both dask_cluster_name and dask_cluster_config_file are set.

static process_graph(task_graph: TaskGraph, *, task_progress_mode: bool = True, **kwargs: Any) → bool

Run a task graph, or report its tasks, by the parsed arguments.

The function prunes the graph to the tasks of task_ids and builds the task cache with TaskCache.create. For the command run, it creates the task executor with create_executor() and runs the graph inside the context of the task executor. For the commands list and status, it prints a table of the task ids and the record type of each task to the standard output, and it runs nothing.

Parameters:

• task_graph (TaskGraph) – The task graph. A run consumes it.

• task_progress_mode (bool) – When true, the task cache is a FileTaskCache over task_status_dir. When false, it is a NoTaskCache. This keyword is separate from the keyword of add_arguments(), so pass the same value to both.

• **kwargs –

  The parsed arguments of add_arguments(). The function reads these keys, and it passes all of them to create_executor():

  command

  run, list or status. None means run.

  task_ids

  The ids of the tasks to keep. The graph keeps the tasks that they depend on too. None means every task.

  task_status_dir

  The task status directory of the FileTaskCache. None, or a missing key, means the current directory.

  force

  When true, the task executor runs every task and writes every record again. The task cache does not change.

  keep_going

  When true, the run continues after a failed task.

  verbose

  When set and not 0, the commands list and status print the exception of each task with an ERROR record. add_arguments() does not add this key. The -v option of VerbosityConfiguration.set_arguments adds it.

Returns:

For run, true when every task succeeds. For list and status, always true.

Return type:

bool

Raises:

• Exception – The error of the first failed task, for run without keep_going.

• ValueError – When command is not run, list or status, or when the graph has no task with one of the task_ids.