qq_lib.cd

Utilities for locating and retrieving the input directory of a job.

This module provides the Cder class, which queries the configured batch system for a job's input directory. The printed path is intended to be consumed by a shell wrapper function that performs the actual directory change.

View Source

 1# Released under MIT License.
 2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
 3
 4"""
 5Utilities for locating and retrieving the input directory of a job.
 6
 7This module provides the `Cder` class, which queries the configured
 8batch system for a job's input directory. The printed path is intended
 9to be consumed by a shell wrapper function that performs the actual directory change.
10"""
11
12from .cder import Cder
13
14__all__ = [
15    "Cder",
16]

class Cder: View Source

21class Cder:
22    """
23    Retrieve and provide the input directory for a specific job in the configured batch system.
24    """
25
26    def __init__(
27        self,
28        BatchSystem: AnyBatchClass,
29        job_id: str,
30    ):
31        """
32        Initialize the Cder instance with a batch system interface and job ID.
33
34        Args:
35            BatchSystem (AnyBatchClass): The batch system which manages the job.
36            job_id (str): Identifier of the job to query.
37        """
38        self._job_id = job_id
39        self._BatchSystem = BatchSystem
40
41    def cd(self) -> str:
42        """
43        Retrieve the input directory for the job and return it as a string.
44
45        Returns:
46            str: Path to the input directory of the job.
47
48        Raises:
49            QQError: If the job does not exist.
50        """
51        path = Cder._get_input_dir_from_job_id(self._BatchSystem, self._job_id)
52        logger.debug(f"Changing directory to '{path}'.")
53        return str(path)
54
55    @staticmethod
56    def _get_input_dir_from_job_id[
57        TBatchJob: BatchJobInterface,
58        TBatchQueue: BatchQueueInterface,
59        TBatchNode: BatchNodeInterface,
60    ](
61        BatchSystem: type[BatchInterface[TBatchJob, TBatchQueue, TBatchNode]],
62        job_id: str,
63    ) -> Path | None:
64        """
65        Query the batch system for the input directory of a job.
66
67        Args:
68            BatchSystem (type[BatchInterface]): The batch system which manages the job.
69            job_id (str): Identifier of the job to query.
70
71        Returns:
72            Path: Path object pointing to the job's input directory.
73
74        Raises:
75            QQError: If the specified job does not exist.
76        """
77        job_info: BatchJobInterface = BatchSystem.get_batch_job(job_id)
78
79        if job_info.is_empty():
80            raise QQError(f"Job '{job_id}' does not exist.")
81
82        if not (input_dir := job_info.get_input_dir()):
83            raise QQError(f"Job '{job_id}' has an unknown input directory.")
84
85        return input_dir

Retrieve and provide the input directory for a specific job in the configured batch system.

Cder(BatchSystem: AnyBatchClass, job_id: str) View Source

26    def __init__(
27        self,
28        BatchSystem: AnyBatchClass,
29        job_id: str,
30    ):
31        """
32        Initialize the Cder instance with a batch system interface and job ID.
33
34        Args:
35            BatchSystem (AnyBatchClass): The batch system which manages the job.
36            job_id (str): Identifier of the job to query.
37        """
38        self._job_id = job_id
39        self._BatchSystem = BatchSystem

Initialize the Cder instance with a batch system interface and job ID.

Arguments:

BatchSystem (AnyBatchClass): The batch system which manages the job.
job_id (str): Identifier of the job to query.

def cd(self) -> str: View Source

41    def cd(self) -> str:
42        """
43        Retrieve the input directory for the job and return it as a string.
44
45        Returns:
46            str: Path to the input directory of the job.
47
48        Raises:
49            QQError: If the job does not exist.
50        """
51        path = Cder._get_input_dir_from_job_id(self._BatchSystem, self._job_id)
52        logger.debug(f"Changing directory to '{path}'.")
53        return str(path)

Retrieve the input directory for the job and return it as a string.

Returns:

str: Path to the input directory of the job.

Raises:

QQError: If the job does not exist.