Process

process

Process module for handling processing steps in flepimop2.

ProcessABC()

Bases: ABC

Abstract base class for flepimop2 processing steps.

Initialize the process with the given configuration.

Source code in src/flepimop2/process/abc.py

def __init__(self) -> None:  # noqa: B027
    """Initialize the process with the given configuration."""

execute(*, dry_run=False)

Execute a processing step.

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, the process will not actually execute but will simulate execution.	False

Source code in src/flepimop2/process/abc.py

def execute(self, *, dry_run: bool = False) -> None:
    """
    Execute a processing step.

    Args:
        dry_run: If True, the process will not actually execute but will simulate
            execution.
    """
    return self._process(dry_run=dry_run)

build(config)

Build a ProcessABC from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any] | ModuleModel	Configuration dictionary. The dict should contain a 'module' key, which will be used to lookup the Process module path. The module will have "flepimop2.process." prepended.	required

Returns:

Name	Type	Description
ProcessABC	ProcessABC	The constructed process object.

Raises:

Type	Description
TypeError	If the built process is not an instance of ProcessABC.

Source code in src/flepimop2/process/abc.py

def build(config: dict[str, Any] | ModuleModel) -> ProcessABC:
    """Build a `ProcessABC` from a configuration dictionary.

    Args:
        config: Configuration dictionary. The dict should contain a
            'module' key, which will be used to lookup the Process module path.
            The module will have "flepimop2.process." prepended.

    Returns:
        ProcessABC: The constructed process object.

    Raises:
        TypeError: If the built process is not an instance of ProcessABC.
    """
    if isinstance(config, ModuleModel):
        config = config.model_dump()
    module = config.pop("module", "shell")
    module_abs = f"flepimop2.process.{module}"
    builder = _load_builder(module_abs)
    process = builder.build(config)
    if not isinstance(process, ProcessABC):
        msg = "The built process is not an instance of ProcessABC."
        raise TypeError(msg)
    return process

abc

Abstract base class for flepimop2 processing steps.

ProcessABC()

Bases: ABC

Abstract base class for flepimop2 processing steps.

Initialize the process with the given configuration.

Source code in src/flepimop2/process/abc.py

def __init__(self) -> None:  # noqa: B027
    """Initialize the process with the given configuration."""

execute(*, dry_run=False)

Execute a processing step.

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, the process will not actually execute but will simulate execution.	False

Source code in src/flepimop2/process/abc.py

def execute(self, *, dry_run: bool = False) -> None:
    """
    Execute a processing step.

    Args:
        dry_run: If True, the process will not actually execute but will simulate
            execution.
    """
    return self._process(dry_run=dry_run)

build(config)

Build a ProcessABC from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any] | ModuleModel	Configuration dictionary. The dict should contain a 'module' key, which will be used to lookup the Process module path. The module will have "flepimop2.process." prepended.	required

Returns:

Name	Type	Description
ProcessABC	ProcessABC	The constructed process object.

Raises:

Type	Description
TypeError	If the built process is not an instance of ProcessABC.

Source code in src/flepimop2/process/abc.py

def build(config: dict[str, Any] | ModuleModel) -> ProcessABC:
    """Build a `ProcessABC` from a configuration dictionary.

    Args:
        config: Configuration dictionary. The dict should contain a
            'module' key, which will be used to lookup the Process module path.
            The module will have "flepimop2.process." prepended.

    Returns:
        ProcessABC: The constructed process object.

    Raises:
        TypeError: If the built process is not an instance of ProcessABC.
    """
    if isinstance(config, ModuleModel):
        config = config.model_dump()
    module = config.pop("module", "shell")
    module_abs = f"flepimop2.process.{module}"
    builder = _load_builder(module_abs)
    process = builder.build(config)
    if not isinstance(process, ProcessABC):
        msg = "The built process is not an instance of ProcessABC."
        raise TypeError(msg)
    return process

shell

Shell process for flepimop2.

ShellProcess()

Bases: ModuleModel, ProcessABC

Shell process for executing commands.

Attributes:

Name	Type	Description
module	Literal['shell']	The module type, fixed to "shell".
command	str	The shell command to execute.
args	list[str]	Arguments to pass to the shell command.

Source code in src/flepimop2/process/abc.py

def __init__(self) -> None:  # noqa: B027
    """Initialize the process with the given configuration."""

build(config)

Build a ShellProcess from a configuration.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary to create a shell process. If module key is defined, it must be "shell". Must contain a 'command' key with the shell command to execute.	required

Returns:

Type	Description
ShellProcess	The ready-to-use ShellProcess instance.

Source code in src/flepimop2/process/shell.py

def build(config: dict[str, Any]) -> ShellProcess:
    """
    Build a [`ShellProcess`][flepimop2.process.shell.ShellProcess] from a configuration.

    Args:
        config: Configuration dictionary to create a shell process. If module key is
            defined, it must be "shell". Must contain a 'command' key with the shell
            command to execute.

    Returns:
        The ready-to-use [`ShellProcess`][flepimop2.process.shell.ShellProcess]
            instance.
    """
    return ShellProcess.model_validate(config)