run — Task for generic command execution

This module defines a few useful functions and classes to embed generic command execution in a depgraph.

Spawning external processes

The RunTask class is the basic building block to run tasks that consist in spawning external processes and waiting for their completion. It makes it possible to execute arbitrary commands. Consider:

>>> from valjean.cosette.run import RunTask
>>> task = RunTask.from_cli(name='say', cli=['echo', 'ni!'])
>>> env_update, status = task.do(env=dict(), config=config) # prints 'ni!'
>>> print(status)
TaskStatus.DONE

The task succeeded, but where is the output of our command?! Note that RunTask captures standard output and standard error and redirects them to files. If you want to see what was printed, you have to look there:

>>> def print_stdout(env_up, name):
...   """A small function to print the stdout of a task."""
...   stdout = env_up[name]['stdout']
...   with open(stdout) as stdout_f:
...     print(stdout_f.read(), end='')
>>> print_stdout(env_update, 'say')
ni!

Note that command is not parsed by a shell. So the following may not do what you expect:

>>> task = RunTask.from_cli(name='want',
...                         cli=['echo', 'We want... ', '&&',
...                              'echo', 'a shrubbery!'])
>>> env_update, status = task.do(env=dict(), config=config)
>>> print_stdout(env_update, 'want')
We want...  && echo a shrubbery!

Indeed, RunTask ran echo only once. If you need to execute several commands, you can wrap them in a shell script and execute it. Alternatively, you can directly invoke the RunTask.from_clis class method:

>>> task = RunTask.from_clis(name='want',
...                          clis=[['echo', '-n', 'We want... '],
...                                ['echo', 'a shrubbery!']])
>>> env_update, status = task.do(env=dict(), config=config)
>>> print_stdout(env_update, 'want')
We want... a shrubbery!

Creating tasks using a factory

When you want to create multiple RunTask objects using the same executable, it can be convenient to use RunTaskFactory. This class can be parametrized to create tasks by specifying the path to the executable once and for all, for instance, and providing the missing arguments later.

The simplest way to create a RunTaskFactory is to use one of the RunTaskFactory.from_executable or RunTaskFactory.from_task class methods. For example, this will create a factory instance that generates RunTask objects for the echo executable:

>>> factory = RunTaskFactory.from_executable('echo', name='echo')

You can use it to generate tasks by invoking the RunTaskFactory.make method:

>>> task = factory.make(name='task', extra_args=['spam'])

This creates an task.echo object (of type RunTask) that executes echo spam when run:

>>> env_up, status = task.do(env={}, config=config)
>>> print_stdout(env_up, 'task.echo')
spam

You can also leave the name parameter out. If you do so, RunTaskFactory will generate a name for you:

>>> task_sausage = factory.make(extra_args=['sausage'])
>>> task_sausage.name
'....echo'

Of course you can generate multiple tasks using the same factory (this is the whole point of RunTaskFactory, really):

>>> task_spam = factory.make(name='task_spam', extra_args=['spam'])
>>> task_eggs = factory.make(name='task_eggs', extra_args=['eggs'])
>>> task_bacon = factory.make(name='task_bacon', extra_args=['bacon'])

You can also specify a few arguments beforehand and provide the rest later:

>>> factory = RunTaskFactory.from_executable('echo', name='echo',
...                                          default_args=['spam'])
>>> task = factory.make(name='task', extra_args=['eggs'])
>>> env_up, status = task.do(env={}, config=config)
>>> print_stdout(env_up, 'task.echo')
spam eggs

Finally, you can parametrize your arguments on arbitrary keywords that will be provided when the task is created:

>>> args = ['{food}', 'with', '{side}']
>>> factory = RunTaskFactory.from_executable('echo', name='echo',
...                                          default_args=args)
>>> task = factory.make(name='task', food='lobster', side='spam')
>>> env_up, status = task.do(env={}, config=config)
>>> print_stdout(env_up, 'task.echo')
lobster with spam

Default values for the parameters may be specified when creating the factory and can be overridden when the task is created:

>>> args = ['{food}', 'with', '{side}']
>>> factory = RunTaskFactory.from_executable('echo', default_args=args,
...                                          side='spam', name='echo')
>>> beans = factory.make(name='baked beans', food='baked beans')
>>> eggs = factory.make(name='eggs', food='eggs',
...                     side='bacon and spam')
>>> env_up, status = beans.do(env={}, config=config)
>>> print_stdout(env_up, 'baked beans.echo')
baked beans with spam
>>> env_up, status = eggs.do(env={}, config=config)
>>> print_stdout(env_up, 'eggs.echo')
eggs with bacon and spam

Note also that you can refer to the environment or the configuration in your command-line arguments:

>>> args = ['{env[side]}']
>>> factory = RunTaskFactory.from_executable('echo', name='echo',
...                                          default_args=args)
>>> task = factory.make(name='task')
>>> env_up, status = task.do(env={'side': 'spam'}, config=config)
>>> print_stdout(env_up, 'task.echo')
spam

Caching

The RunTaskFactory class caches generated tasks under the hood. Repeated calls to RunTaskFactory.make from the same factory with the same arguments will result in the same task:

>>> factory = RunTaskFactory.from_executable('echo', name='echo')
>>> task_sausage = factory.make(extra_args=['sausage'])
>>> task_sausage_again = factory.make(extra_args=['sausage'])
>>> task_sausage is task_sausage_again
True

If you instantiate another factory, the caching mechanism is defeated:

>>> other_factory = RunTaskFactory.from_executable('echo', name='echo')
>>> task_sausage_other = other_factory.make(extra_args=['sausage'])
>>> task_sausage is task_sausage_other
False

Module API

valjean.cosette.run.run(clis, stdout, stderr, **subprocess_args)

Run the given command lines and capture their stdout/stderr.

Execution stops at the first failure (result value != 0).

Parameters:

• clis (list) – The list of commands to execute.

• stdout (file object) – File handle to capture the stdout stream.

• stderr (file object) – File handle to capture the stderr stream.

• subprocess_args (dict) – Parameters to be passed to subprocess.call.

valjean.cosette.run.make_cap_paths(base_path)

Construct filenames to capture stdout and stderr.

class valjean.cosette.run.RunTask(name, clis_closure, *, deps=None, soft_deps=None, **subprocess_args)

Task that executes the specified shell commands and waits for their completion.

classmethod from_cli(name, cli, **kwargs)

Create a RunTask from a single command line.

Use the RunTask.from_clis method if you want to run several commands in a row.

Parameters:

• name (str) – The name of this task.

• cli (list) – The command line to be executed, as a list of strings. The first element of the list is the command and the following ones are its arguments.

• kwargs – Any other keyword arguments will be passed on to the constructor (see RunTask.__init__).

classmethod from_clis(name, clis, **kwargs)

Create a RunTask from a list of command lines.

Parameters:

• name (str) – The name of this task.

• clis (list) – The command lines to be executed, as a list of lists of strings. The first element of each sub-list is the command and the following ones are its arguments.

• kwargs – Any other keyword arguments will be passed on to the constructor (see RunTask.__init__).

__init__(name, clis_closure, *, deps=None, soft_deps=None, **subprocess_args)

Initialize this task from a list of command lines.

The clis_closure argument must be a closure. It will be invoked at execution time as:

clis_closure(env, config)

and it must return the command lines to be executed, as a list of lists of strings.

Parameters:

• name (str) – The name of this task.

• clis_closure (list) – A closure to generate the command lines.

• subprocess_args (dict) – Any remaining options will be passed to the subprocess.Popen constructor.

• deps (list(Task) or None) – The dependencies for this task (see Task.__init__ for the format), or None.

• soft_deps (list(Task) or None) – The dependencies for this task (see Task.__init__ for the format), or None.

run_task(clis_closure, name, **subprocess_args)

Execute the specified command and wait for its completion.

On completion, this method proposes the following updates to the environment:

env[task.name]['clis'] = clis
env[task.name]['return_codes'] = return_codes
env[task.name]['elapsed_time'] = wallclock_time
env[task.name]['stdout'] = stdout
env[task.name]['stderr'] = stderr

Here clis is the list of command lines that were executed, return_codes is the list of return codes of the executed commands, and elapsed_time is the time the whole list took. The keys stdout and stderr hold the paths to the files containing respectively the captured standard output and standard error streams.

Parameters:

• env (Env) – The task environment.

• config (Config or None) – The configuration object.

Returns:

The proposed environment update.

class valjean.cosette.run.RunTaskFactory(make_closure, *, deps=None, soft_deps=None, name, **kwargs)

Create multiple tasks from the same executable without even breaking a sweat.

classmethod from_task(task, *, relative_path, name=None, default_args=None, **kwargs)

This class method creates a RunTaskFactory from a Task. The command to be executed must appear in the output_dir directory of the given task; its relative location must be specified with the relative_path argument.

This method can be used, among other things, to create a RunTaskFactory from a CheckoutTask or a BuildTask.

Parameters:

• task (Task) – The Task object.

• relative_path (str) – The path to the executable, relative to the the output directory (output_dir) of the task.

• name (str or None) – a unique identifier for this factory. The factory name is used to produce unique names for the generated tasks. If None is given, the factory name will be constructed by hashing the other arguments.

• default_args – see RunTaskFactory.from_executable.

• kwargs – Any remaining arguments will be passed to the __init__.

classmethod from_executable(path, name=None, default_args=None, **kwargs)

This class method creates a RunTaskFactory from the path to an existing executable.

Parameters:

• path (str) – the path to the executable.

• name (str or None) – a unique identifier for this factory. If None is given, the factory will use a hash of the other arguments.

• default_args (list(str) or None) – The list of arguments that will be passed to the executed command by RunTask. It may contain expressions understood by Python’s format mini-language, such as {foo}. These expressions can be filled when the task is generated by passing appropriate keyword arguments to the make method.

• kwargs – Any remaining arguments will be passed to the __init__.

__init__(make_closure, *, deps=None, soft_deps=None, name, **kwargs)

make(*, name=None, extra_args=None, subprocess_args=None, deps=None, soft_deps=None, **kwargs)

Create a RunTask object.

Parameters:

• name (str or None) – the name of the task to be generated, as a string. If None is passed, RunTaskFactory will generate a name by hashing the contents of all the other arguments.

• extra_args (list or None) – A list of additional arguments that will be appended at the end of the command line.

• subprocess_args (dict or None) – A dictionary of arguments for the call to the subprocess.Popen constructor.

• deps (list(Task) or None) – A list of dependencies for the generated task. Note that factories built with from_task automatically inject dependencies on the given task.

• soft_deps (list(Task) or None) – A list of soft dependencies for the generated task.

• kwargs – Any remaining keyword arguments will be used to format the command line before execution. The environment and the configuration are available at formatting time as env and config, respectively.

copy()

Return a copy of this object.