Utilities

Utilities#

`requires` and `inherits`#

b2luigi provides two functions to help you with the task creation. The first one is b2luigi.requires(), which is a decorator that allows you to specify the dependencies of a task easily. It is similar to the requires method in luigi, but it allows you to specify the dependencies in a more flexible way. The second one is b2luigi.inherits(), which is a decorator that allows you to inherit the parameters of a task easily. It is similar to the inherits method in luigi, but it allows you to inherit without specific parameters.

b2luigi.requires(*tasks_to_require, **kwargs)[source]#

This “hack” copies the luigi.requires functionality, except that we allow for additional kwarg arguments when called.

It can be used to require a certain task, but with some variables already set, e.g.

class TaskA(b2luigi.Task):
    some_parameter = b2luigi.IntParameter()
    some_other_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("test.txt")

@b2luigi.requires(TaskA, some_parameter=3)
class TaskB(b2luigi.Task):
    another_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("out.dat")

TaskB will not require TaskA, where some_parameter is already set to 3. This also means, that TaskB only has the parameters another_parameter and some_other_parameter (because some_parameter is already fixed).

It is also possible to require multiple tasks, e.g.

class TaskA(b2luigi.Task):
    some_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("test.txt")

class TaskB(b2luigi.Task):
    some_other_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("test.txt")

@b2luigi.requires(TaskA, TaskB)
class TaskC(b2luigi.Task):
    another_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("out.dat")

b2luigi.inherits(*tasks_to_inherit, **kwargs)[source]#

This copies the luigi.inherits functionality but allows specifying parameters you don’t want to inherit.

It can e.g. be used in tasks that merge the output of the tasks they require. These merger tasks don’t need the parameter they resolve anymore but should keep the same order of parameters, therefore simplifying the directory structure created by b2luigi.Task.add_to_output().

Usage can be similar to this:

class TaskA(b2luigi.Task):
    some_parameter = b2luigi.IntParameter()
    some_other_parameter = b2luigi.IntParameter()

    def output(self):
        yield self.add_to_output("test.txt")

@b2luigi.inherits(TaskA, without='some_other_parameter')
class TaskB(b2luigi.Task):
    another_parameter = b2luigi.IntParameter()

    def requires(self):
        for my_other_parameter in range(10):
            yield self.clone(TaskA, some_other_parameter=my_other_parameter)

    def run(self):
        # somehow merge the output of TaskA to create "out.dat"
        pass

    def output(self):
        yield self.add_to_output("out.dat")

Parameters:: without – Either a string or a collection of strings

See also: b2luigi.requires which extends luigi.requires.

Core Utilities#

b2luigi.core.utils.SYNTHETIC_TASK_MODULE = 'TaskClasses'#: Module name assigned to the user’s task file when the CLI imports it. Classes defined in the task file carry this as __module__; classes imported into it keep their real module. The batch encoding (create_cmd_from_task) and the CLI task index both branch on it.

b2luigi.core.utils.product_dict(**kwargs: Any) → Iterator[Dict[str, Any]][source]#

Cross-product the given parameters and return a list of dictionaries.

Example

>>> list(product_dict(arg_1=[1, 2], arg_2=[3, 4]))
[{'arg_1': 1, 'arg_2': 3}, {'arg_1': 1, 'arg_2': 4}, {'arg_1': 2, 'arg_2': 3}, {'arg_1': 2, 'arg_2': 4}]

The thus produced list can directly be used as inputs for a required tasks:

def requires(self):
    for args in product_dict(arg_1=[1, 2], arg_2=[3, 4]):
        yield some_task(**args)

Parameters:: kwargs (Any) – Each keyword argument should be an iterable
Returns:: A list of kwargs where each list of input keyword arguments is cross-multiplied with every other.
Return type:: Iterator[Dict[str, Any]]

b2luigi.core.utils.fill_kwargs_with_lists(**kwargs: Any) → Dict[str, List[Any]][source]#

Return the kwargs with each value mapped to [value] if not a list already.

Example: .. code-block:: python

>>> fill_kwargs_with_lists(arg_1=[1, 2], arg_2=3)
{'arg_1': [1, 2], 'arg_2': [3]}

Parameters:: kwargs – The input keyword arguments
Returns:: Same as kwargs, but each value mapped to a list if not a list already
Return type:: Dict[str, List[Any]]

b2luigi.core.utils.flatten_to_file_paths(inputs: Iterable[FileSystemTarget]) → Dict[str, List[str]][source]#

Take in a dict of lists of luigi targets and replace each luigi target by its corresponding path. For dicts, it will replace the value as well as the key. The key will however only by the basename of the path.

Parameters:: inputs – A dict of lists of luigi targets
Returns:: A dict with the keys replaced by the basename of the targets and the values by the full path
Return type:: Dict[str, List[str]]

b2luigi.core.utils.flatten_to_dict(inputs: Iterable[Any]) → Dict[Any, Any][source]#

Return a whatever input structure into a dictionary. If it is a dict already, return this. If is is an iterable of dict or dict-like objects, return the merged dictionary. All non-dict values will be turned into a dictionary with value -> {value: value}

Example: .. code-block:: python

>>> flatten_to_dict([{"a": 1, "b": 2}, {"c": 3}, "d"])
{'a': 1, 'b': 2, 'c': 3, 'd': 'd'}

Parameters:: inputs (Iterable[Any]) – The input structure
Returns:: A dict constructed as described above.
Return type:: Dict[Any, Any]

b2luigi.core.utils.flatten_to_dict_of_lists(inputs: Iterable[Any]) → Dict[Any, List][source]#

Flattens a nested iterable structure into a dictionary of lists.

This function takes an iterable of potentially nested structures, flattens it, and converts it into a dictionary where each key maps to a list of values associated with that key.

Parameters:: inputs (Iterable[Any]) – The input iterable containing nested structures.
Returns:: A dictionary where keys are derived from the input and values are lists of corresponding items.
Return type:: Dict[Any, List]

b2luigi.core.utils.task_iterator(task, only_non_complete=False)[source]#

Iterates through a task and its dependencies in a directed acyclic graph (DAG), ensuring that each task is yielded only once.

Parameters:

task – The root task to start iterating from. This task should have methods complete() to check if the task is complete and deps() to retrieve its dependencies.
only_non_complete (bool, optional) – If True, only tasks that are not complete (as determined by the complete() method) will be yielded. Defaults to False.

Yields:

task –

Each unique task in the DAG, starting from the given root task and: including its dependencies.

Notes

The function uses a cache (already_seen_tasks) to ensure that tasks are not yielded multiple times, even if they are dependencies of multiple parent tasks in the DAG.
The iteration is performed recursively using a nested helper function.

b2luigi.core.utils.get_all_output_files_in_tree(root_module, key=None)[source]#

Recursively retrieves all output files from tasks in a given module tree.

This function iterates through all tasks in the specified root module, collects their output files, and organizes them into a dictionary. If a specific key is provided, it returns the output files corresponding to that key.

Parameters:

root_module (module) – The root module containing tasks to iterate over.
key (str, optional) – A specific key to filter the output files. If provided, only the output files corresponding to this key will be returned.

Returns:

A dictionary where keys are file identifiers and values are lists of dictionaries containing:

exists (bool): Whether the file exists.

parameters (dict): Serialized parameters of the task.

file_name (str): Absolute path to the file.

Return type:

dict

Raises:

KeyError – If the specified key is not found in the output files.

Notes

The function uses task_iterator to traverse tasks in the module tree.
Output files are flattened and converted to absolute file paths.

b2luigi.core.utils.get_serialized_parameters(task, need_hidden=False)[source]#

Retrieve a string-typed ordered dictionary of significant parameters in the format key=value.

This function iterates over the parameters of a given task, filters out non-significant parameters, and serializes the significant ones. If a parameter has a custom serialization method (serialize_hashed), it will use that; otherwise, it defaults to the standard serialize method.

Parameters:

task – An object representing a task, which must implement the get_params method to retrieve its parameters and their metadata.

Returns:

An ordered dictionary where keys are parameter names and values: are their serialized representations.

Return type:

collections.OrderedDict

b2luigi.core.utils.create_output_file_name(task, base_filename: str, result_dir: str | None = None, need_hidden: bool = False) → str[source]#

Generates an output file path based on the task’s parameters, a base filename, and an optional result directory.

Parameters:

task – The task object containing parameters to serialize and use in the output path.
base_filename (str) – The base name of the output file.
result_dir (Optional[str]) – The directory where the output file should be saved. If not provided, it defaults to the result_dir setting.

Returns:

The full path to the output file.

Return type:

str

Raises:

ValueError – If any parameter value contains a path separator or cannot be interpreted as a valid directory name.

Notes

The function ensures that the result directory is evaluated relative to the current executed file.
The parameter separator and whether to include parameter names in the output path are configurable via settings.
If use_parameter_name_in_output is enabled, the output path includes parameter names and values; otherwise, only parameter values are used.
If parameter_separator is set to a non-empty string, it will be used to separate parameter names and values in the output path.

b2luigi.core.utils.get_log_file_dir(task)[source]#

Determines the directory where log files for a given task should be stored.

If the task has a custom method get_log_file_dir, it will use that method to retrieve the log file directory. Otherwise, it constructs the log file directory path based on the task’s settings and family.

Parameters:: task – The task object for which the log file directory is being determined.
Returns:: The path to the log file directory for the given task.
Return type:: str

b2luigi.core.utils.get_task_file_dir(task)[source]#

Determines the directory path for a given task’s output files.

If the task has a method get_task_file_dir, it will use that method to retrieve the directory path. Otherwise, it generates the directory path using the create_output_file_name function and the task’s family name.

Parameters:: task – An object representing the task. It is expected to have a get_task_file_dir method or a get_task_family method.
Returns:: The directory path for the task’s output files.
Return type:: str

b2luigi.core.utils.get_filename()[source]#

Returns the absolute path of the task-definitions file.

Resolution order:

``B2LUIGI_TASK_FILE`` environment variable — set this when the task file is not the script being executed directly (e.g. when using the b2luigi CLI with python -m b2luigi).
``__main__.__file__`` when it is a .py file — the classic python tasks.py invocation; the executed script IS the task file.
``os.getcwd()/tasks.py`` fallback — used when the process is started via a CLI binary (e.g. the installed b2luigi script, which has no .py extension). The current working directory is the project root in this case, and dirname(cwd/tasks.py) == cwd gives callers the correct base directory for resolving relative paths.

Returns:: The absolute path of the task-definitions file.
Return type:: str
Raises:: AttributeError – If __main__.__file__ is not set (e.g. in a Jupyter notebook) and B2LUIGI_TASK_FILE is also not set.

b2luigi.core.utils.map_folder(input_folder)[source]#

Maps a relative folder path to an absolute path based on the location of the current script.

If the input folder path is already absolute, it is returned as-is. Otherwise, the function determines the directory of the current script and joins it with the relative input folder path to produce an absolute path.

Parameters:: input_folder (str) – The folder path to map. Can be either absolute or relative.
Returns:: The absolute path corresponding to the input folder.
Return type:: str
Raises:: AttributeError – If the current script location cannot be determined, typically when running in an interactive shell (e.g., Jupyter Notebook). In such cases, the user is advised to provide absolute paths in their settings.

b2luigi.core.utils.on_failure(task, _)[source]#

Handles the failure of a task by generating an explanation message, printing it to stdout, and returning it to be sent back to the scheduler.

Parameters:: task – The task instance that failed.
Returns:: A detailed explanation of the failure, including the task ID, parameters, and the location of the log files.
Return type:: str

b2luigi.core.utils.add_on_failure_function(task)[source]#

Assigns a custom failure handler to the given task.

This function dynamically binds the on_failure method to the provided task object.

Parameters:: task – The task object to which the on_failure method will be attached.

b2luigi.core.utils.create_cmd_from_task(task)[source]#

Constructs a command-line argument list to execute a task on a batch worker node.

Branches on the internal __batch_runner_use_cli setting:

New CLI mode (True): emits the Typer-based batch-runner invocation. By default it runs the module directly through a Python interpreter:
```
<executable_prefix> <executable> -m <batch_runner_cli> batch-runner
    --classname TaskFamily
    --param key=value …
    [--task-file /abs/path/to/tasks.py]
    <task_cmd_additional_args>
```
If executable_is_entrypoint is True, the -m <batch_runner_cli> step is skipped and executable is invoked directly as the b2luigi entrypoint instead:
```
<executable_prefix> <executable> batch-runner
    --classname TaskFamily
    --param key=value …
    [--task-file /abs/path/to/tasks.py]
    <task_cmd_additional_args>
```
This avoids baking an absolute, submission-host-specific interpreter path into the executable wrapper, which breaks when the batch execution environment has a different filesystem view than the submission host (e.g. a container batch universe). Instead, executable (default: [<batch_runner_cli>], e.g. ["b2luigi"]) is resolved via PATH at execution time.

Set automatically by b2luigi.cli.utils.process_task_instance() — do not set __batch_runner_use_cli manually.
Old mode (False, default): emits the legacy argparse invocation:
```
<executable_prefix> <executable> [filename] --batch-runner --task-id TaskFamily_id
    <task_cmd_additional_args>
```
filename is omitted when the add_filename_to_cmd setting is False. executable_is_entrypoint has no effect here: this mode executes an arbitrary user script file, which always requires a Python interpreter and has no entrypoint-only equivalent.

The batch_runner_cli setting (default: "b2luigi") lets projects that ship their own CLI built on top of b2luigi override the module name (-m mode) or the default entrypoint name (executable_is_entrypoint=True mode, when executable is otherwise unset):

set_setting("batch_runner_cli", "flare")

executable defaults depend on mode: [sys.executable] in -m mode, [<batch_runner_cli>] in entrypoint mode. executable_is_entrypoint itself defaults to True only when executable has not been explicitly set anywhere in the settings cascade — if you already override executable (e.g. to a specific interpreter), the -m default is preserved so existing setups keep working unchanged. When executable_is_entrypoint=True and executable is explicitly set, batch_runner_cli is silently ignored (there’s no -m step for it to configure).

Parameters:

task – An object representing the task for which the command is being created.

Returns:

A list of strings representing the command-line arguments, intended to be flattened into a shell context (e.g. " ".join(...) embedded in a generated shell script, as b2luigi.core.executable.create_executable_wrapper() does), not passed directly to subprocess.Popen without shell=True. The b2luigi-generated --param key=value and --task-file tokens are individually shlex.quote-d so they survive that flattening intact even when a parameter value contains spaces, brackets, or quotes. User-supplied settings — executable_prefix, executable, and task_cmd_additional_args — are passed through verbatim, by design: they may themselves already contain multiple shell words (e.g. executable_prefix=["nice", "-n", "10"]) that must not be quoted into a single token.

In CLI mode, the --classname token is module-qualified (pkg.mod.ClassName) for classes not defined in the task file itself, so the worker can reconstruct classes that are not attributes of the task file’s namespace (e.g. a task imported transitively by a module the task file imports). Classes defined in the task file are still sent bare: their synthetic module name is not importable, but the worker imports the task file anyway and finds them there. As a side effect, tasks using luigi’s task_namespace — previously encoded as namespace.ClassName via get_task_family(), which the worker’s plain getattr could never resolve — are now encoded by class provenance and resolve correctly.

Return type:

list

Raises:

ValueError – If task_cmd_additional_args, executable_prefix, or executable is a plain string rather than a list.

b2luigi.core.utils.create_output_dirs(task)[source]#

Creates the necessary output directories for a given task.

This function takes a task object, retrieves its outputs, and ensures that the directories required for those outputs exist by calling the makedirs method on each target.

Parameters:: task – The task object whose outputs need directories to be created.

b2luigi.core.utils.get_filled_params(task)[source]#

Retrieve a dictionary of parameter names and their corresponding values from a given task.

Parameters:

task – An object representing a task, which must have a get_params method that returns an iterable of parameter name and metadata pairs, and attributes corresponding to the parameter names.

Returns:

A dictionary containing parameter names as keys and their: respective values as values.

Return type:

dict

b2luigi.core.utils.is_subdir(path, parent_dir)[source]#

Determines if a given path is a subdirectory of a specified parent directory.

Parameters:

path (str) – The path to check.
parent_dir (str) – The parent directory to compare against.

Returns:

True if the path is a subdirectory of the parent directory, False otherwise.

Return type:

bool

Note

Both path and parent_dir are converted to their absolute paths before comparison.

b2luigi.core.utils.get_apptainer_or_singularity(task=None)[source]#

Determines the command to use for containerization, prioritizing Apptainer over Singularity. If neither is available, raises a ValueError.

The function first checks if a custom command is set via the apptainer_cmd setting. If not, it checks for the availability of the apptainer or singularity commands in the system’s PATH.

Parameters:

task (optional) – An optional task object that may be used to retrieve task-specific settings.

Returns:

The command to use for containerization, either “apptainer” or: ”singularity”.

Return type:

str

Raises:

ValueError – If neither Apptainer nor Singularity is available on the system and no custom command is set.

b2luigi.core.utils.create_apptainer_command(command, task=None)[source]#

Constructs a command to execute within an Apptainer (or Singularity) container.

This function generates a command string that sets up the necessary environment and mounts for running a given command inside an Apptainer container. It ensures that required settings are provided and validates compatibility with the batch system.

Parameters:

command (str) – The command to be executed inside the Apptainer container.
task (optional) – An optional task object or identifier used to retrieve task-specific settings.

Returns:

A clean, unquoted argv list representing the full Apptainer execution command. No element carries its own shell quoting; quoting (or passing the list directly to subprocess.Popen without shell=True) is the caller’s responsibility.

Return type:

list

Raises:

ValueError – If the env_script is not provided.
ValueError – If the batch system is gbasf2, as Apptainer is not supported for this batch system.

Notes

apptainer_image is retrieved from the task settings.
apptainer_additional_params is used to specify additional parameters for the Apptainer command. Accepts either a single string (word-split via shlex.split(), e.g. "--cleanenv --nv") or a list of strings (used verbatim, e.g. ["--cleanenv", "--nv"]).
apptainer_mounts is used to specify additional mount points for the Apptainer command. Expecting a list of strings.
apptainer_mount_defaults determines whether to include default mount points (e.g., result directory and log file directory).

b2luigi.core.utils.get_luigi_logger()[source]#

Retrieve the logger instance used by luigi.

Returns:: The logger instance for “luigi-interface”.
Return type:: logging.Logger

Executable#

b2luigi.core.executable.create_executable_wrapper(task)[source]#

Creates a bash script wrapper to execute a task with the appropriate environment and settings. The wrapper script sets up the working directory, environment variables, and optionally uses an Apptainer image for execution.

Parameters:: task – The task containing configuration and settings.
Returns:: The file path to the generated executable wrapper script.
Return type:: str

The wrapper script performs the following steps:

Changes to the working_dir directory.
Sets up the environment:
- Sources env_script if provided.
- Overrides environment variables based on task or settings from env.
Constructs the command to execute the task, see create_cmd_from_task.
Executes the command:
- If an Apptainer image is specified in apptainer_image, runs the command within the image.
- Otherwise, executes the command directly with exec.
Writes the generated script to a file and makes it executable.

b2luigi.core.executable.run_task_remote(task)[source]#

Executes a given task remotely by creating an executable script and running it via a subprocess call. The standard output and error streams are redirected to log files.

Parameters:: task – The task to be executed
Raises:: RuntimeError – If the subprocess call returns a non-zero exit code, indicating a failure during execution.

Side Effects:

Creates a directory for log files if it does not already exist.
Writes the standard output and error of the subprocess execution to separate log files.

Utilities

Contents

Utilities#

requires and inherits#

Core Utilities#

Executable#

`requires` and `inherits`#