Launchers¶

import kloch.launchers

Getters¶

kloch.launchers.get_available_launchers_classes(launcher_plugins: List[str] | None = None) → List[Type[BaseLauncher]]¶

Get all list of available launcher classes that are registred.

Parameters:: launcher_plugins – list of python module names to load plugin launcher from.

kloch.launchers.get_launcher_class(name: str) → Type[BaseLauncher] | None¶: Get the launcher class which match the given unique name.

kloch.launchers.get_available_launchers_serialized_classes(launcher_plugins: List[str] | None = None) → List[Type[BaseLauncherSerialized]]¶

Get all list of available serialized launcher classes that are registred.

Parameters:: launcher_plugins – list of python module names to load plugin launcher from.

kloch.launchers.get_launcher_serialized_class(identifier: str) → Type[BaseLauncherSerialized] | None¶: Get the serialized launcher class which match the given identifier.

Plugins¶

kloch.launchers.check_launcher_plugins(launcher_plugins: List[str])¶

Raise an exception if any of the launchers extracted from plugin system are invalid.

Parameters:: launcher_plugins – list of modules name to extract launcher from.

kloch.launchers.is_launcher_plugin(launcher: Type[BaseLauncher] | Type[BaseLauncherSerialized]) → bool¶: Return True if the given launcher is an external plugin else False if builtin.

Serialized¶

class kloch.launchers.LauncherSerializedDict¶

Bases: MergeableDict

A list of launchers instance serialized as a dict structure.

The dict is expected to have the following root structure:

{"manager_name1": {...}, "manager_name2": {...}, ...}

The dict structure include tokens that need to be resolved and indicate how to merge 2 LauncherSerializedDict instances together. See MergeableDict for the full documentation on tokens.

to_serialized_list() → LauncherSerializedList¶

Convert the dict structure to a list of BaseLauncherSerialized instances.

Returns:: deepcopied dict structure as list of instances.

class kloch.launchers.LauncherSerializedList(iterable=(), /)¶

Bases: List[BaseLauncherSerialized]

to_dict() → Dict[str, Dict]¶

Convert the list to a builtin dict structure (noc ustom class used).

Useful for serialization or to convert to a LauncherSerializedDict instance.

with_base_merged() → LauncherSerializedList¶

Get a copy of this instance with the .base launcher merged with the other launchers.

(This implies the returned instance does NOT have a .base key anymore)

Returns:: new instance with deepcopied structure.

BaseLauncher¶

class kloch.launchers.BaseLauncher(environ: ~typing.Dict[str, str] = <factory>, command: ~typing.List[str] = <factory>, cwd: str | None = None)¶

Bases: object

An “abstract” dataclass that describe how to start a software environment session.

command: List[str]¶

Arbitrary list of command line arguments to call at the end of the launcher execution.

The developer is reponsible of honoring the field usage in its launcher implementation.

cwd: str | None = None¶

Current working directory.

The developer is reponsible of honoring the field usage in its launcher implementation.

environ: Dict[str, str]¶

Mapping of environment variables to set when starting the environment.

The developer is reponsible of honoring the field usage in its launcher implementation.

abstract execute(tmpdir: Path, command: List[str] | None = None) → int¶

Start the given environment and execute this python session.

Optionally execute the given command in the environment.

Parameters:

tmpdir – filesystem path to an existing temporary directory
command – optional list of command line arguments

Returns:

The exit code of the execution. 0 if successfull, else imply failure.

classmethod from_dict(src_dict: Dict[str, Any]) → BaseLauncher¶: Generate an instance from a python dict object with a specific structure.

name: ClassVar[str] = '.base'¶: A unique name among all subclasses.

required_fields: ClassVar[List[str]] = []¶

List of dataclass field that are required to have a non-None value when instancing.

Note that your subclass must have the default field value set to None for this to work.

Example:

@dataclasses.dataclass
class DemoLauncher(BaseLauncher):
    # override the BaseLauncher.environ field to make it required
    environ: Dict[str, str] = None

    required_fields = ["environ"]

to_dict() → Dict[str, Any]¶: Convert the instance to a python dict object.

class kloch.launchers.BaseLauncherSerialized¶

Bases: MergeableDict

A BaseLauncher instance as a serialized dict object.

The dict structure might be invalid and might need resolving, before being unserialized to a valid BaseLauncher instance.

fields¶: alias of BaseLauncherFields

identifier: str = '.base'¶

value pair.

Type:: An unique name among serialized launcher that can be use in a dict key

resolved() → Dict¶: Modify the dict structure, so it can be unserialized properly.

source¶: alias of BaseLauncher

unserialize() → BaseLauncher¶: Convert the dict to a valid instance.

abstract validate()¶

Ensure the dict structure can be resolved to a valid structure than can be unserialized.

Raise an exeception on any issue.

class kloch.launchers.BaseLauncherFields(environ: Dict[str, str | List[str]] = 'environ', merge_system_environ: bool = 'merge_system_environ', command: List[str] = 'command', cwd: str = 'cwd')¶

A mapping of ‘dataclass field name’: ‘corresponding key name in serialized dict’.

Where the field name is the variable name. This implies you can have a different variable string value than the variable name.

Example: config_params="configParams" is valid.

This class is never instanced and all attributes are used at class level.

classmethod iterate() → List[Field]¶: Return all the fields defined on this dataclass

BaseLauncher Subclasses¶

class kloch.launchers.SystemLauncher(environ: ~typing.Dict[str, str] = <factory>, command: ~typing.List[str] = <factory>, cwd: str | None = None)¶

Bases: BaseLauncher

A minimal launcher that just start a subprocess with the given command.

command: List[str]¶

Arbitrary list of command line arguments to call at the end of the launcher execution.

The developer is reponsible of honoring the field usage in its launcher implementation.

cwd: str | None = None¶

Current working directory.

The developer is reponsible of honoring the field usage in its launcher implementation.

environ: Dict[str, str]¶

Mapping of environment variables to set when starting the environment.

The developer is reponsible of honoring the field usage in its launcher implementation.

execute(tmpdir: Path, command: List[str] | None = None)¶: Just call subprocess.run.

classmethod from_dict(src_dict: Dict[str, Any]) → BaseLauncher¶: Generate an instance from a python dict object with a specific structure.

name: ClassVar[str] = 'system'¶: A unique name among all subclasses.

required_fields: ClassVar[List[str]] = []¶

List of dataclass field that are required to have a non-None value when instancing.

Note that your subclass must have the default field value set to None for this to work.

Example:

@dataclasses.dataclass
class DemoLauncher(BaseLauncher):
    # override the BaseLauncher.environ field to make it required
    environ: Dict[str, str] = None

    required_fields = ["environ"]

to_dict() → Dict[str, Any]¶: Convert the instance to a python dict object.

class kloch.launchers.SystemLauncherSerialized¶

Bases: BaseLauncherSerialized

fields¶: alias of SystemLauncherFields

identifier: str = 'system'¶

value pair.

Type:: An unique name among serialized launcher that can be use in a dict key

source¶: alias of SystemLauncher

unserialize() → SystemLauncher¶: Convert the dict to a valid instance.

validate()¶

Ensure the dict structure can be resolved to a valid structure than can be unserialized.

Raise an exeception on any issue.

class kloch.launchers.PythonLauncher(environ: ~typing.Dict[str, str] = <factory>, command: ~typing.List[str] = <factory>, cwd: str | None = None, python_file: str | None = None)¶

Bases: BaseLauncher

A launcher that execute the given python file with kloch’s own interpreter.

command: List[str]¶

Arbitrary list of command line arguments to call at the end of the launcher execution.

The developer is reponsible of honoring the field usage in its launcher implementation.

cwd: str | None = None¶

Current working directory.

The developer is reponsible of honoring the field usage in its launcher implementation.

environ: Dict[str, str]¶

Mapping of environment variables to set when starting the environment.

The developer is reponsible of honoring the field usage in its launcher implementation.

execute(tmpdir: Path, command: List[str] | None = None)¶: Just call subprocess.run with sys.executable + the file path

classmethod from_dict(src_dict: Dict[str, Any]) → BaseLauncher¶: Generate an instance from a python dict object with a specific structure.

name: ClassVar[str] = '@python'¶: A unique name among all subclasses.

python_file: str = None¶: Filesystem path to an existing python file.

required_fields: ClassVar[List[str]] = ['python_file']¶

List of dataclass field that are required to have a non-None value when instancing.

Note that your subclass must have the default field value set to None for this to work.

Example:

@dataclasses.dataclass
class DemoLauncher(BaseLauncher):
    # override the BaseLauncher.environ field to make it required
    environ: Dict[str, str] = None

    required_fields = ["environ"]

to_dict() → Dict[str, Any]¶: Convert the instance to a python dict object.

class kloch.launchers.PythonLauncherSerialized¶

Bases: BaseLauncherSerialized

fields¶: alias of PythonLauncherFields

identifier: str = '@python'¶

value pair.

Type:: An unique name among serialized launcher that can be use in a dict key

resolved() → Dict¶: Modify the dict structure, so it can be unserialized properly.

source¶: alias of PythonLauncher

unserialize() → PythonLauncher¶: Convert the dict to a valid instance.

validate()¶

Ensure the dict structure can be resolved to a valid structure than can be unserialized.

Raise an exeception on any issue.