Skip to content

endpoint_operations

create_package_dir_pip(*, task_pkg, create=True)

Create venv folder for a task package and return corresponding Path object

Source code in fractal_server/tasks/v2/endpoint_operations.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def create_package_dir_pip(
    *,
    task_pkg: _TaskCollectPip,
    create: bool = True,
) -> Path:
    """
    Create venv folder for a task package and return corresponding Path object
    """
    settings = Inject(get_settings)
    user = FRACTAL_PUBLIC_TASK_SUBDIR
    if task_pkg.package_version is None:
        raise ValueError(
            f"Cannot create venv folder for package `{task_pkg.package}` "
            "with `version=None`."
        )
    package_dir = f"{task_pkg.package_name}{task_pkg.package_version}"
    venv_path = settings.FRACTAL_TASKS_DIR / user / package_dir
    if create:
        venv_path.mkdir(exist_ok=False, parents=True)
    return venv_path

download_package(*, task_pkg, dest) async

Download package to destination and return wheel-file path.

Source code in fractal_server/tasks/v2/endpoint_operations.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def download_package(
    *,
    task_pkg: _TaskCollectPip,
    dest: Union[str, Path],
) -> Path:
    """
    Download package to destination and return wheel-file path.
    """
    interpreter = get_python_interpreter_v2(
        python_version=task_pkg.python_version
    )
    pip = f"{interpreter} -m pip"
    if task_pkg.package_version is None:
        package_and_version = f"{task_pkg.package_name}"
    else:
        package_and_version = (
            f"{task_pkg.package_name}=={task_pkg.package_version}"
        )
    cmd = f"{pip} download --no-deps {package_and_version} -d {dest}"
    stdout = await execute_command(command=cmd, cwd=Path("."))
    pkg_file = next(
        line.split()[-1] for line in stdout.split("\n") if "Saved" in line
    )
    return Path(pkg_file)

inspect_package(path, logger_name=None)

Inspect task package to extract version and manifest

Note that this only works with wheel files, which have a well-defined dist-info section. If we need to generalize to to tar.gz archives, we would need to go and look for PKG-INFO.

Parameters:

Name Type Description Default
path Path

Path of the package wheel file.

 required
logger_name Optional[str]
None

Returns:

Type Description
dict[Literal['pkg_version', 'pkg_manifest'], str]

A dictionary with keys pkg_version and pkg_manifest.
Source code in fractal_server/tasks/v2/endpoint_operations.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def inspect_package(
    path: Path, logger_name: Optional[str] = None
) -> dict[Literal["pkg_version", "pkg_manifest"], str]:
    """
    Inspect task package to extract version and manifest

    Note that this only works with wheel files, which have a well-defined
    dist-info section. If we need to generalize to to tar.gz archives, we would
    need to go and look for `PKG-INFO`.

    Args:
        path: Path of the package wheel file.
        logger_name:

    Returns:
        A dictionary with keys `pkg_version` and `pkg_manifest`.
    """

    logger = get_logger(logger_name)

    if not path.as_posix().endswith(".whl"):
        raise ValueError(
            "Only wheel packages are supported in Fractal "
            f"(given {path.name})."
        )

    # Extract package name and version from wheel filename
    _info = _parse_wheel_filename(wheel_filename=path.name)
    pkg_version = _info["version"]

    # Read and validate task manifest
    logger.debug(f"Now reading manifest for {path.as_posix()}")
    with ZipFile(path) as wheel:
        pkg_manifest = _load_manifest_from_wheel(
            path, wheel, logger_name=logger_name
        )
    logger.debug("Manifest read correctly.")

    info = dict(
        pkg_version=pkg_version,
        pkg_manifest=pkg_manifest,
    )
    return info