resolve the cluster paths through the cluster view

airflow_src/dags/impl/handler_impl.py

@@ -2,7 +2,7 @@
 
 import logging
 import re
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 
 from airflow.exceptions import AirflowFailException, AirflowSkipException
 from airflow.models import TaskInstance
@@ -59,9 +59,10 @@
     DDA_FLAG_IN_RAW_FILE_NAME,
 )
 from shared.path_layout import get_raw_file_folder_rel_path
+from shared.path_views import CLUSTER_VIEW, Locations
 from shared.settings_scope_resolver import resolve_scoped_settings
 from shared.validation import FORBIDDEN_RAW_FILE_NAME_CHARACTERS_PATTERN
-from shared.yamlsettings import YamlKeys, get_path, is_s3_upload_enabled
+from shared.yamlsettings import is_s3_upload_enabled
 
 # special mode that does not copy (e.g. because another instance handles it)
 # point locations.backup.absolute_path to the folder where the files can be picked up for quanting
@@ -322,9 +323,12 @@ def _handle_file_copying(
     return copied_files
 
 
-def get_backup_base_path(raw_file: RawFile) -> Path:
+# TODO: move
+def get_backup_base_path(raw_file: RawFile) -> PurePosixPath:
     """Get the backup base path for the given raw file, e.g. /fs/pool/backup/test2/2025_07 ."""
-    return get_path(YamlKeys.Locations.BACKUP) / get_raw_file_folder_rel_path(raw_file)
+    return CLUSTER_VIEW.resolve(
+        Locations.BACKUP, get_raw_file_folder_rel_path(raw_file)
+    )
 
 
 def _verify_copied_files(

airflow_src/dags/impl/processor_impl.py

@@ -3,7 +3,7 @@
 import json
 import logging
 from collections import defaultdict
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 
 from airflow.exceptions import AirflowFailException, AirflowSkipException
 from airflow.models import TaskInstance
@@ -56,9 +56,9 @@
 from shared.db.models import RawFile, RawFileStatus, Settings, get_created_at_year_month
 from shared.keys import SoftwareTypes
 from shared.path_layout import get_output_folder_rel_path, get_raw_file_rel_path
+from shared.path_views import CLUSTER_VIEW, Locations
 from shared.settings_scope_resolver import resolve_scoped_settings
 from shared.validation import check_for_malicious_content
-from shared.yamlsettings import YamlKeys, get_path
 
 
 class QuantingFailedNewErrorException(AirflowFailException):
@@ -120,7 +120,7 @@ def prepare_job(raw_file_id: str, settings_id: str) -> dict:
     settings = get_settings_by_id(settings_id)
 
     relative_raw_file_path = get_raw_file_rel_path(raw_file)
-    raw_file_path = get_path(YamlKeys.Locations.BACKUP) / relative_raw_file_path
+    raw_file_path = CLUSTER_VIEW.resolve(Locations.BACKUP, relative_raw_file_path)
 
     internal_output_path = get_internal_output_path_for_raw_file(
         raw_file, software_type=settings.software_type
@@ -166,12 +166,12 @@ def _find_next_free_run_suffix(base_path: Path) -> str:
 def _create_quanting_env(
     settings: Settings,
     raw_file: RawFile,
-    raw_file_path: Path,
+    raw_file_path: PurePosixPath,
     relative_raw_file_path: Path,
     output_path_suffix: str = "",
 ) -> QuantingEnv:
     """Create a quanting environment from settings."""
-    settings_path = get_path(YamlKeys.Locations.SETTINGS) / settings.name
+    settings_path = CLUSTER_VIEW.resolve(Locations.SETTINGS, settings.name)
 
     relative_output_path = get_output_folder_rel_path(
         raw_file, software_type=settings.software_type
@@ -181,7 +181,7 @@ def _create_quanting_env(
             relative_output_path.name + output_path_suffix
         )
 
-    output_path = get_path(YamlKeys.Locations.OUTPUT) / relative_output_path
+    output_path = CLUSTER_VIEW.resolve(Locations.OUTPUT, relative_output_path)
     internal_output_path = get_internal_output_path() / relative_output_path
 
     substituted_params = _substitute_config_params(
@@ -236,11 +236,11 @@ def _create_quanting_env(
 def _substitute_config_params(  # noqa: PLR0913 Too many arguments
     raw_file_id: str,
     relative_output_path: Path,
-    output_path: Path,
+    output_path: PurePosixPath,
     relative_raw_file_path: Path,
-    raw_file_path: Path,
+    raw_file_path: PurePosixPath,
     settings: Settings,
-    settings_path: Path,
+    settings_path: PurePosixPath,
     num_threads: int,
     project_id: str,
 ) -> str:
@@ -265,8 +265,7 @@ def _substitute_config_params(  # noqa: PLR0913 Too many arguments
 
 def _prepare_custom_command(settings: Settings, substituted_params: str) -> str:
     """Prepare the custom command for the quanting job."""
-    software_base_path = get_path(YamlKeys.Locations.SOFTWARE)
-    software_path = str(software_base_path / settings.software)
+    software_path = str(CLUSTER_VIEW.resolve(Locations.SOFTWARE, settings.software))
 
     custom_command = f"{software_path} {substituted_params}"
     logging.info(f"Custom command for quanting: {custom_command}")

airflow_src/plugins/jobs/job_handler.py

@@ -11,7 +11,8 @@
 from common.quanting_env import QuantingEnv
 
 from shared.keys import JobEngines
-from shared.yamlsettings import YamlKeys, get_host_mounts_path, get_path
+from shared.path_views import CLUSTER_VIEW, Locations
+from shared.yamlsettings import get_host_mounts_path
 
 
 def _get_job_handler(engine: str) -> "JobHandler":
@@ -20,7 +21,7 @@ def _get_job_handler(engine: str) -> "JobHandler":
         from jobs.slurm_ssh_job_handler import SlurmSSHJobHandler
 
         logging.info("Using SlurmSSHJobHandler")
-        return SlurmSSHJobHandler(get_path(YamlKeys.Locations.SLURM))
+        return SlurmSSHJobHandler(CLUSTER_VIEW.resolve(Locations.SLURM))
 
     if engine == JobEngines.DOCKER:
         try:

airflow_src/plugins/jobs/slurm_ssh_job_handler.py

@@ -2,7 +2,7 @@
 
 import logging
 from datetime import datetime
-from pathlib import Path
+from pathlib import PurePosixPath
 
 from airflow.exceptions import AirflowFailException
 from common.constants import (
@@ -19,7 +19,7 @@
 class SlurmSSHJobHandler(JobHandler):
     """Implementation of JobHandler that executes commands on a Slurm cluster via SSH."""
 
-    def __init__(self, cluster_base_dir: Path):
+    def __init__(self, cluster_base_dir: PurePosixPath):
         """Initialize the Slurm job handler.
 
         Args:

airflow_src/tests/common/test_settings.py

@@ -1,13 +1,10 @@
 """Tests for the settings module."""
 
-from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 from common.settings import get_instrument_settings
 
-from shared.yamlsettings import get_path
-
 
 def test_get_instrument_settings_returns_setting_for_existing_instrument_and_key() -> (
     None
@@ -30,41 +27,6 @@ def test_get_instrument_settings_raises_key_error_for_non_existing_key() -> None
         get_instrument_settings("instrument1", "key2")
 
 
-def test_get_path_returns_setting_for_existing_instrument_and_key() -> None:
-    """Test that correct path is returned."""
-    with (
-        patch(
-            "shared.yamlsettings.YAMLSETTINGS",
-            {"locations": {"backup": {"absolute_path": "some_path"}}},
-        ),
-    ):
-        assert get_path("backup") == Path("some_path")
-
-
-def test_get_path_returns_setting_raises_key_error_for_non_exisiting_key_1() -> None:
-    """Test that a KeyError is raised if the key does not exist."""
-    with (
-        patch(
-            "shared.yamlsettings.YAMLSETTINGS",
-            {"locations": {"backup": {"absolute_path": "some_path"}}},
-        ),
-        pytest.raises(KeyError),
-    ):
-        get_path("Xbackup")
-
-
-def test_get_path_returns_setting_raises_key_error_for_non_exisiting_key_2() -> None:
-    """Test that a KeyError is raised if the key does not exist."""
-    with (
-        patch(
-            "shared.yamlsettings.YAMLSETTINGS",
-            {"locations": {"backup": {"Xabsolute_path": "some_path"}}},
-        ),
-        pytest.raises(KeyError),
-    ):
-        get_path("backup")
-
-
 def test_get_instrument_settings_raises_key_error_for_non_existing_instrument() -> None:
     """Test that a KeyError is raised if the instrument does not exist in the instrument settings."""
     with (

airflow_src/tests/helpers.py

@@ -6,29 +6,18 @@
 from unittest.mock import patch
 
 from shared.path_views import CLUSTER_VIEW
-from shared.yamlsettings import YAMLSETTINGS, YamlKeys
 
 
 @contextmanager
 def yaml_locations(**paths: str) -> Iterator[None]:
-    """Override the `locations` section of the yaml settings, e.g. `yaml_locations(slurm="/path/to/slurm")`.
+    """Override the locations of `CLUSTER_VIEW`, e.g. `yaml_locations(slurm="/path/to/slurm")`.
 
-    Patches the settings that `get_path()` reads and the contents of `CLUSTER_VIEW`, rather than
-    either of them at its import site, so tests stay valid when code moves between modules.
+    Patches the contents of the view object rather than its name, so the patch reaches every
+    import site and tests stay valid when code moves between modules.
     """
-    with (
-        patch.dict(
-            YAMLSETTINGS,
-            {
-                YamlKeys.LOCATIONS: {
-                    key: {YamlKeys.ABSOLUTE_PATH: path} for key, path in paths.items()
-                }
-            },
-        ),
-        patch.object(
-            CLUSTER_VIEW,
-            "_locations",
-            {key: PurePosixPath(path) for key, path in paths.items()},
-        ),
+    with patch.object(
+        CLUSTER_VIEW,
+        "_locations",
+        {key: PurePosixPath(path) for key, path in paths.items()},
     ):
         yield

shared/yamlsettings.py

@@ -104,22 +104,6 @@ def load_alphakraken_yaml(cls) -> dict[str, dict[str, Any]]:
 )
 
 
-def get_path(path_key: str) -> Path:
-    """Get a certain path from the yaml settings."""
-    path = (
-        YAMLSETTINGS.get(YamlKeys.LOCATIONS, {})  # type: ignore[possibly-unbound-attribute]
-        .get(path_key, {})
-        .get(YamlKeys.ABSOLUTE_PATH)
-    )
-
-    if path is None:
-        raise KeyError(
-            f"Key `{YamlKeys.LOCATIONS}.{path_key}` or `{YamlKeys.LOCATIONS}.{path_key}.{YamlKeys.ABSOLUTE_PATH}` not found in alphakraken.yaml."
-        )
-
-    return Path(path)
-
-
 def get_host_mounts_path() -> Path:
     """Get the path of the mounts folder as seen by the docker host (not by the containers)."""
     path = (

webapp/pages_/settings.py

@@ -308,7 +308,7 @@ def display_settings(
             "label": "Executable*",
             "max_chars": 64,
             "placeholder": "e.g. 'custom-software/custom-executable1.2.3'",
-            # "help": f"Path to executable, relative to `{get_path(YamlKeys.Locations.SOFTWARE)}/`. Ask an administrator to add the executable to the software folder. "
+            # TODO: reimplement using the actual software path, cf. CLUSTER_VIEW
             "help": "Path to executable, relative to the software folder. Ask an administrator to add the executable to the software folder. "
             f"If something that is in the `$PATH` should be executed, it needs to be wrapped by a shell script located in the software folder. "
             f"For the `{JobEngines.DOCKER}` execution engine, this is a docker image name instead, e.g. `alphakraken-msqc`. "

webapp/tests/pages_/test_overview.py

@@ -5,20 +5,8 @@
 from unittest.mock import MagicMock, patch
 
 import pandas as pd
-
-import shared.yamlsettings
-
-
-def mock_get_path(path_key: str) -> Path:
-    """Mock get_path to return dummy paths for testing."""
-    del path_key  # Unused parameter
-    return Path("/some/path")
-
-
-shared.yamlsettings.get_path = mock_get_path  # type: ignore[invalid-assignment]
-
-from service.session_state import SessionStateKeys  # noqa: E402
-from streamlit.testing.v1 import AppTest  # noqa: E402]
+from service.session_state import SessionStateKeys
+from streamlit.testing.v1 import AppTest
 
 PAGES_FOLDER = Path(__file__).parent / Path("../../pages_")
 

webapp/tests/pages_/test_status.py

@@ -5,19 +5,7 @@
 from unittest.mock import MagicMock, patch
 
 import pandas as pd
-
-import shared.yamlsettings
-
-
-def mock_get_path(path_key: str) -> Path:
-    """Mock get_path to return dummy paths for testing."""
-    del path_key  # Unused parameter
-    return Path("/some/path")
-
-
-shared.yamlsettings.get_path = mock_get_path  # type: ignore[invalid-assignment]
-
-from streamlit.testing.v1 import AppTest  # noqa: E402
+from streamlit.testing.v1 import AppTest
 
 PAGES_FOLDER = Path(__file__).parent / Path("../../pages_")
 

webapp/tests/service/test_components.py

@@ -7,16 +7,14 @@
 
 import pandas as pd
 import pytest
-
-with patch("shared.yamlsettings.get_path") as p:  # TODO: ugly!
-    from service.components import (
-        _get_color,
-        display_status,
-        get_full_backup_path,
-        highlight_status_cell,
-        show_date_select,
-        show_filter,
-    )
+from service.components import (
+    _get_color,
+    display_status,
+    get_full_backup_path,
+    highlight_status_cell,
+    show_date_select,
+    show_filter,
+)
 
 
 @pytest.mark.parametrize(