Merge branch 'main' into 1945-replace-the-two-args_schema-query-param…

…eters-in-get-apiv2task-with-just-one

CHANGELOG.md

@@ -1,8 +1,16 @@
 **Note**: Numbers like (\#1234) point to closed Pull Requests on the fractal-server repository.
 
-# 2.7.2 (Unreleased)
-
-* API:
+# 2.8.0
+
+* Task group
+    * Add `pinned_package_versions_string` property (\#1980).
+* Task collection
+    * Now both the local and SSH versions of the task collection use the bash templates (\#1980).
+    * Update task-collections database logs incrementally (\#1980).
+    * Support pinned-package versions for SSH task collection (\#1980).
+    * Now `pip install` uses `--no-cache` (\#1980).
+* API
+    * Deprecate the `verbose` query parameter in `GET /api/v2/task/collect/{state_id}/` (\#1980).
     * Combine the `args_schema_parallel` and `args_schema_non_parallel` query parameters in `GET /api/v2/task/` into a single parameter `args_schema` (\#1998).
 
 # 2.7.1

docs/development.md

@@ -40,13 +40,6 @@ its dependencies, including development dependencies.
 Note that to run commands from within this environment you should prepend them
 with `poetry run`, as in `poetry run fractalctl start`.
 
-To install Fractal Server with some additional extras, use the [`-E`
-option](https://python-poetry.org/docs/pyproject/#extras), as in
-```console
-poetry install -E postgres
-poetry install --all-extras
-```
-
 It may sometimes be useful to use a different Python interpreter from the one
 installed in your system. To this end we suggest using
 [pyenv](https://github.com/pyenv/pyenv). In the project folder, invoking

fractal_server/app/models/v2/task.py

@@ -103,3 +103,18 @@ def pip_install_string(self) -> str:
                     f"{self.pkg_name=}, {self.wheel_path=}, {self.version=}."
                 )
             return f"{self.pkg_name}{extras}=={self.version}"
+
+    @property
+    def pinned_package_versions_string(self) -> str:
+        """
+        Prepare string to be used in `python -m pip install`.
+        """
+        if self.pinned_package_versions is None:
+            return ""
+        output = " ".join(
+            [
+                f"{key}=={value}"
+                for key, value in self.pinned_package_versions.items()
+            ]
+        )
+        return output

fractal_server/app/routes/api/v2/task_collection.py

@@ -24,23 +24,22 @@
 from ....schemas.v2 import TaskCollectPipV2
 from ....schemas.v2 import TaskGroupCreateV2
 from ...aux.validate_user_settings import validate_user_settings
+from ._aux_functions_task_collection import get_package_version_from_pypi
 from ._aux_functions_tasks import _get_valid_user_group_id
 from ._aux_functions_tasks import _verify_non_duplication_group_constraint
 from ._aux_functions_tasks import _verify_non_duplication_user_constraint
 from fractal_server.app.models import UserOAuth
 from fractal_server.app.routes.auth import current_active_user
 from fractal_server.app.routes.auth import current_active_verified_user
 from fractal_server.app.schemas.v2 import TaskGroupV2OriginEnum
-from fractal_server.tasks.utils import _normalize_package_name
-from fractal_server.tasks.utils import get_collection_log_v2
-from fractal_server.tasks.v2.background_operations import (
-    background_collect_pip,
+from fractal_server.tasks.v2.collection_local import (
+    collect_package_local,
 )
-from fractal_server.tasks.v2.endpoint_operations import (
-    get_package_version_from_pypi,
+from fractal_server.tasks.v2.utils_package_names import _parse_wheel_filename
+from fractal_server.tasks.v2.utils_package_names import normalize_package_name
+from fractal_server.tasks.v2.utils_python_interpreter import (
+    get_python_interpreter_v2,
 )
-from fractal_server.tasks.v2.utils import _parse_wheel_filename
-from fractal_server.tasks.v2.utils import get_python_interpreter_v2
 
 router = APIRouter()
 
@@ -118,14 +117,14 @@ async def collect_tasks_pip(
                     f"Original error: {str(e)}",
                 ),
             )
-        task_group_attrs["pkg_name"] = _normalize_package_name(
+        task_group_attrs["pkg_name"] = normalize_package_name(
             wheel_info["distribution"]
         )
         task_group_attrs["version"] = wheel_info["version"]
         task_group_attrs["origin"] = TaskGroupV2OriginEnum.WHEELFILE
     else:
         pkg_name = task_collect.package
-        task_group_attrs["pkg_name"] = _normalize_package_name(pkg_name)
+        task_group_attrs["pkg_name"] = normalize_package_name(pkg_name)
         task_group_attrs["origin"] = TaskGroupV2OriginEnum.PYPI
         latest_version = await get_package_version_from_pypi(
             task_collect.package,
@@ -249,8 +248,8 @@ async def collect_tasks_pip(
     if settings.FRACTAL_RUNNER_BACKEND == "slurm_ssh":
         # SSH task collection
 
-        from fractal_server.tasks.v2.background_operations_ssh import (
-            background_collect_pip_ssh,
+        from fractal_server.tasks.v2.collection_ssh import (
+            collect_package_ssh,
         )
 
         # User appropriate FractalSSH object
@@ -263,7 +262,7 @@ async def collect_tasks_pip(
         fractal_ssh = fractal_ssh_list.get(**ssh_credentials)
 
         background_tasks.add_task(
-            background_collect_pip_ssh,
+            collect_package_ssh,
             state_id=state.id,
             task_group=task_group,
             fractal_ssh=fractal_ssh,
@@ -273,7 +272,7 @@ async def collect_tasks_pip(
     else:
         # Local task collection
         background_tasks.add_task(
-            background_collect_pip,
+            collect_package_local,
             state_id=state.id,
             task_group=task_group,
         )
@@ -296,42 +295,15 @@ async def collect_tasks_pip(
 async def check_collection_status(
     state_id: int,
     user: UserOAuth = Depends(current_active_user),
-    verbose: bool = False,
     db: AsyncSession = Depends(get_async_db),
 ) -> CollectionStateReadV2:  # State[TaskCollectStatus]
     """
     Check status of background task collection
     """
-
-    logger = set_logger(logger_name="check_collection_status")
-    logger.debug(f"Querying state for state.id={state_id}")
     state = await db.get(CollectionStateV2, state_id)
-    if not state:
-        await db.close()
+    if state is None:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"No task collection info with id={state_id}",
         )
-
-    settings = Inject(get_settings)
-    if settings.FRACTAL_RUNNER_BACKEND == "slurm_ssh":
-        # FIXME SSH: add logic for when data.state["log"] is empty
-        pass
-    else:
-        # Non-SSH mode
-        # In some cases (i.e. a successful or ongoing task collection),
-        # state.data["log"] is not set; if so, we collect the current logs.
-        if verbose and not state.data.get("log"):
-            if "path" not in state.data.keys():
-                await db.close()
-                raise HTTPException(
-                    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-                    detail=(
-                        f"No 'path' in CollectionStateV2[{state_id}].data"
-                    ),
-                )
-            state.data["log"] = get_collection_log_v2(Path(state.data["path"]))
-
-    reset_logger_handlers(logger)
-    await db.close()
     return state

fractal_server/app/routes/api/v2/task_collection_custom.py

@@ -27,12 +27,12 @@
 from fractal_server.logger import set_logger
 from fractal_server.string_tools import validate_cmd
 from fractal_server.syringe import Inject
-from fractal_server.tasks.v2.background_operations import (
-    _prepare_tasks_metadata,
-)
 from fractal_server.tasks.v2.database_operations import (
     create_db_tasks_and_update_task_group,
 )
+from fractal_server.tasks.v2.utils_background import (
+    _prepare_tasks_metadata,
+)
 
 router = APIRouter()
 

fractal_server/app/schemas/v2/task_collection.py

@@ -10,7 +10,6 @@
 from pydantic import root_validator
 from pydantic import validator
 
-from .._validators import valdictkeys
 from .._validators import valstr
 from fractal_server.app.schemas._validators import valutc
 from fractal_server.app.schemas.v2 import ManifestV2
@@ -62,13 +61,30 @@ class TaskCollectPipV2(BaseModel, extra=Extra.forbid):
     _package_version = validator("package_version", allow_reuse=True)(
         valstr("package_version")
     )
-    _pinned_package_versions = validator(
-        "pinned_package_versions", allow_reuse=True
-    )(valdictkeys("pinned_package_versions"))
     _package_extras = validator("package_extras", allow_reuse=True)(
         valstr("package_extras")
     )
 
+    @validator("pinned_package_versions")
+    def pinned_package_validator(cls, value):
+        if value is None:
+            return value
+        old_keys = list(value.keys())
+        new_keys = [
+            valstr(f"pinned_package_versions[{key}]")(key) for key in old_keys
+        ]
+        if len(new_keys) != len(set(new_keys)):
+            raise ValueError(
+                f"Dictionary contains multiple identical keys: {value}."
+            )
+        for old_key, new_key in zip(old_keys, new_keys):
+            if new_key != old_key:
+                value[new_key] = value.pop(old_key)
+        for pkg, version in value.items():
+            validate_cmd(pkg)
+            validate_cmd(version)
+        return value
+
     @validator("package")
     def package_validator(cls, value):
         if "/" in value or value.endswith(".whl"):

fractal_server/data_migrations/old/2_7_0.py

@@ -25,7 +25,7 @@
 )
 from fractal_server.app.security import FRACTAL_DEFAULT_GROUP_NAME
 from fractal_server.data_migrations.tools import _check_current_version
-from fractal_server.tasks.utils import _normalize_package_name
+from fractal_server.tasks.v2.utils_package_names import normalize_package_name
 from fractal_server.utils import get_timestamp
 
 logger = logging.getLogger("fix_db")
@@ -171,7 +171,7 @@ def prepare_task_groups(
                 python_version,
                 name,
             ) = source_fields
-            pkg_name = _normalize_package_name(pkg_name)
+            pkg_name = normalize_package_name(pkg_name)
             task_group_key = ":".join(
                 [pkg_name, version, extras, python_version]
             )

fractal_server/tasks/utils.py

@@ -1,4 +1,3 @@
-import re
 from pathlib import Path
 
 from fractal_server.config import get_settings
@@ -42,38 +41,8 @@ def get_collection_log_v1(path: Path) -> str:
     return log
 
 
-def get_collection_log_v2(path: Path) -> str:
-    log_path = get_log_path(path)
-    log = log_path.open().read()
-    return log
-
-
 def get_collection_freeze_v1(venv_path: Path) -> str:
     package_path = get_absolute_venv_path_v1(venv_path)
     freeze_path = get_freeze_path(package_path)
     freeze = freeze_path.open().read()
     return freeze
-
-
-def get_collection_freeze_v2(path: Path) -> str:
-    freeze_path = get_freeze_path(path)
-    freeze = freeze_path.open().read()
-    return freeze
-
-
-def _normalize_package_name(name: str) -> str:
-    """
-    Implement PyPa specifications for package-name normalization
-
-    The name should be lowercased with all runs of the characters `.`, `-`,
-    or `_` replaced with a single `-` character. This can be implemented in
-    Python with the re module.
-    (https://packaging.python.org/en/latest/specifications/name-normalization)
-
-    Args:
-        name: The non-normalized package name.
-
-    Returns:
-        The normalized package name.
-    """
-    return re.sub(r"[-_.]+", "-", name).lower()

fractal_server/tasks/v1/background_operations.py

@@ -7,10 +7,10 @@
 from shutil import rmtree as shell_rmtree
 
 from ...string_tools import slugify_task_name_for_source_v1
-from ..utils import _normalize_package_name
 from ..utils import get_collection_log_v1
 from ..utils import get_collection_path
 from ..utils import get_log_path
+from ..v2.utils_package_names import normalize_package_name
 from ._TaskCollectPip import _TaskCollectPip
 from .utils import _init_venv_v1
 from fractal_server.app.db import DBSyncSession
@@ -23,7 +23,7 @@
 from fractal_server.logger import close_logger
 from fractal_server.logger import get_logger
 from fractal_server.logger import set_logger
-from fractal_server.utils import execute_command
+from fractal_server.utils import execute_command_async
 
 
 async def _pip_install(
@@ -60,12 +60,12 @@ async def _pip_install(
     cmd_install = f"{pip} install {pip_install_str}"
     cmd_inspect = f"{pip} show {task_pkg.package}"
 
-    await execute_command(
+    await execute_command_async(
         cwd=venv_path,
         command=f"{pip} install --upgrade pip",
         logger_name=logger_name,
     )
-    await execute_command(
+    await execute_command_async(
         cwd=venv_path, command=cmd_install, logger_name=logger_name
     )
     if task_pkg.pinned_package_versions:
@@ -82,7 +82,7 @@ async def _pip_install(
                 "Preliminary check: verify that "
                 f"{pinned_pkg_version} is already installed"
             )
-            stdout_inspect = await execute_command(
+            stdout_inspect = await execute_command_async(
                 cwd=venv_path,
                 command=f"{pip} show {pinned_pkg_name}",
                 logger_name=logger_name,
@@ -99,7 +99,7 @@ async def _pip_install(
                     f"({pinned_pkg_version}); "
                     f"install version {pinned_pkg_version}."
                 )
-                await execute_command(
+                await execute_command_async(
                     cwd=venv_path,
                     command=(
                         f"{pip} install "
@@ -114,7 +114,7 @@ async def _pip_install(
                 )
 
     # Extract package installation path from `pip show`
-    stdout_inspect = await execute_command(
+    stdout_inspect = await execute_command_async(
         cwd=venv_path, command=cmd_inspect, logger_name=logger_name
     )
 
@@ -165,8 +165,8 @@ async def _create_venv_install_package(
     """
 
     # Normalize package name
-    task_pkg.package_name = _normalize_package_name(task_pkg.package_name)
-    task_pkg.package = _normalize_package_name(task_pkg.package)
+    task_pkg.package_name = normalize_package_name(task_pkg.package_name)
+    task_pkg.package = normalize_package_name(task_pkg.package)
 
     python_bin = await _init_venv_v1(
         path=path,
@@ -192,8 +192,8 @@ async def create_package_environment_pip(
     logger = get_logger(logger_name)
 
     # Normalize package name
-    task_pkg.package_name = _normalize_package_name(task_pkg.package_name)
-    task_pkg.package = _normalize_package_name(task_pkg.package)
+    task_pkg.package_name = normalize_package_name(task_pkg.package_name)
+    task_pkg.package = normalize_package_name(task_pkg.package)
 
     # Only proceed if package, version and manifest attributes are set
     task_pkg.check()