_utils

_customize_and_run_template(template_filename, replacements, script_dir, logger_name, prefix)

Customize one of the template bash scripts.

PARAMETER	DESCRIPTION
template_filename	Filename of the template file (ends with ".sh"). TYPE: str
replacements	Dictionary of replacements. TYPE: set[tuple[str, str]]
script_dir	Local folder where the script will be placed. TYPE: str
prefix	Prefix for the script filename. TYPE: int

Source code in fractal_server/tasks/v2/local/_utils.py

def _customize_and_run_template(
    template_filename: str,
    replacements: set[tuple[str, str]],
    script_dir: str,
    logger_name: str,
    prefix: int,
) -> str:
    """
    Customize one of the template bash scripts.

    Args:
        template_filename: Filename of the template file (ends with ".sh").
        replacements: Dictionary of replacements.
        script_dir: Local folder where the script will be placed.
        prefix: Prefix for the script filename.
    """
    logger = get_logger(logger_name=logger_name)
    logger.debug(f"_customize_and_run_template {template_filename} - START")

    # Prepare name and path of script
    if not template_filename.endswith(".sh"):
        raise ValueError(
            f"Invalid {template_filename=} (it must end with '.sh')."
        )

    script_filename = f"{prefix}_{template_filename}"
    script_path_local = Path(script_dir) / script_filename
    # Read template
    customize_template(
        template_name=template_filename,
        replacements=replacements,
        script_path=script_path_local,
    )
    cmd = f"bash {script_path_local}"
    logger.debug(f"Now run '{cmd}' ")
    stdout = execute_command_sync(command=cmd, logger_name=logger_name)
    logger.debug(f"_customize_and_run_template {template_filename} - END")
    return stdout

check_task_files_exist(task_list)

Check that the modules listed in task commands point to existing files.

PARAMETER	DESCRIPTION
task_list	TYPE: list[TaskCreate]

Source code in fractal_server/tasks/v2/local/_utils.py

def check_task_files_exist(task_list: list[TaskCreate]) -> None:
    """
    Check that the modules listed in task commands point to existing files.

    Args:
        task_list:
    """

    for _task in task_list:
        if _task.command_non_parallel is not None:
            _task_path = _task.command_non_parallel.split()[-1]
            if not Path(_task_path).exists():
                raise FileNotFoundError(
                    f"Task `{_task.name}` has `command_non_parallel` "
                    f"pointing to missing file `{_task_path}`."
                )
        if _task.command_parallel is not None:
            _task_path = _task.command_parallel.split()[-1]
            if not Path(_task_path).exists():
                raise FileNotFoundError(
                    f"Task `{_task.name}` has `command_parallel` "
                    f"pointing to missing file `{_task_path}`."
                )

edit_pyproject_toml_in_place_local(pyproject_toml_path, resource)

Wrapper of simplify_pyproject_toml, with I/O.

Source code in fractal_server/tasks/v2/local/_utils.py

def edit_pyproject_toml_in_place_local(
    pyproject_toml_path: Path,
    resource: Resource,
) -> None:
    """
    Wrapper of `simplify_pyproject_toml`, with I/O.
    """

    # Read `pyproject.toml`
    with pyproject_toml_path.open() as f:
        pyproject_contents = f.read()

    # Simplify contents
    new_pyproject_contents = simplify_pyproject_toml(
        original_toml_string=pyproject_contents,
        pixi_environment=resource.tasks_pixi_config["DEFAULT_ENVIRONMENT"],
        pixi_platform=resource.tasks_pixi_config["DEFAULT_PLATFORM"],
    )
    # Write new `pyproject.toml`
    with pyproject_toml_path.open("w") as f:
        f.write(new_pyproject_contents)
    logger.debug(
        f"Replaced local {pyproject_toml_path.as_posix()} "
        "with simplified version."
    )

rmtree_nofail(*, folder_path, logger_name)

Wrap a shutil.rmtree with try/except and logs.

PARAMETER	DESCRIPTION
folder_path	Folder to be deleted. TYPE: str
logger_name	Logger name. TYPE: str

Source code in fractal_server/tasks/v2/local/_utils.py

def rmtree_nofail(
    *,
    folder_path: str,
    logger_name: str,
) -> None:
    """
    Wrap a `shutil.rmtree` with try/except and logs.

    Args:
        folder_path: Folder to be deleted.
        logger_name: Logger name.
    """
    logger = get_logger(logger_name)
    try:
        logger.info(f"Now delete folder {folder_path}")
        shutil.rmtree(folder_path)
        logger.info(f"Deleted folder {folder_path}")
    except Exception as e:
        logger.error(f"Removing folder failed.\nOriginal error:\n{str(e)}")