Launchers¶

import kloch.launchers

Contexts¶

class kloch.launchers.LauncherContext(platform: LauncherPlatform | None = None, user: str | None = None)¶

Collection of system-contextual values in which a profile is being parsed.

You can compare 2 context using the equal comparator however be aware that a None value (implies unset) will be considered equal to any other value.

Example:

ctx1 = LauncherContext(platform=LauncherPlatform.linux, user="demo")
ctx2 = LauncherContext(platform=LauncherPlatform.linux, user=None)
assert ctx1 == ctx2

classmethod create_from_system()¶: Fill the instance with values retrieved from the current system context.

platform: LauncherPlatform | None = None¶: name of the operating system

user: str | None = None¶: name of the current user

class kloch.launchers.LauncherPlatform(value)¶

Operating Systems that a launcher may support.

classmethod current()¶: Get the member corresponding to the current runtime operating system.

darwin = 2¶

is_linux() → bool¶

is_mac() → bool¶

is_windows() → bool¶

linux = 1¶

windows = 4¶

Getters¶

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

Collect all the launchers classes available.

This is simply the given plugin launchers + builtin launchers.

Parameters:: plugins – collection of launcher loaded from plugins.

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

Collect all the serialized launchers classes available.

This is simply the given plugin launchers + builtin launchers.

Parameters:: plugins – collection of launcher loaded from plugins

Plugins¶

class kloch.launchers.LoadedPluginsLaunchers(launchers: List[T], missed: Dict[str, str], given: List[str])¶

kloch.launchers.check_launcher_plugins(plugins: LoadedPluginsLaunchers[Type[BaseLauncherSerialized]]) → List[Exception]¶

Return any issue/error the given launcher may have.

This function is not supposed to raise if a launcher is invalid.

Parameters:: plugins – collection of plugin launcher loaded from external modules.
Returns:: list of errors found in given loaded plugins, empty if no errors.

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.

get_filtered_context(context: LauncherContext) → LauncherSerializedDict¶

Remove all launchers that doesn’t match the given context.

Returns:: a new deepcopied instance with possibly lesser keys.

to_serialized_list(launcher_classes: List[Type[BaseLauncherSerialized]]) → LauncherSerializedList¶

Convert the dict structure to a list of BaseLauncherSerialized instances.

Parameters:: launcher_classes – list of launchers classes that can be possibly stored in this serialized dict.
Raises:: ValueError – If a launcher is serialized in this instance but is unknown.
Returns:: deepcopied dict structure as list of instances.

with_context_resolved() → LauncherSerializedDict¶: Merge all the same launchers with different contexts to a single launcher.

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, priority: int = 0)¶

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.

priority: int = 0¶: How much you should privilege this launcher to be used over other launchers.

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', priority: int = 'priority')¶

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, priority: int = 0, subprocess_kwargs: dict = <factory>)¶

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.

priority: int = 0¶: How much you should privilege this launcher to be used over other launchers.

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"]

subprocess_kwargs: dict¶: Mapping of kwargs to pass to python’s ‘subprocess.run’ call.

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, priority: int = 0, 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.

priority: int = 0¶: How much you should privilege this launcher to be used over other launchers.

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.