qq_lib.properties.transfer_mode

  1# Released under MIT License.
  2# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
  3
  4import re
  5from abc import ABC, abstractmethod
  6from dataclasses import dataclass
  7
  8from qq_lib.core.error import QQError
  9
 10
 11@dataclass(frozen=True)
 12class TransferMode(ABC):
 13    """
 14    Specifies when job data should be transferred from the working directory to
 15    the input directory or archived from the working directory.
 16    """
 17
 18    @classmethod
 19    def from_str(cls, s: str) -> "TransferMode":
 20        """
 21        Convert a string to the corresponding TransferMode.
 22
 23        Args:
 24            s (str): String representation of the transfer mode.
 25
 26        Returns:
 27            TransferMode variant.
 28
 29        Raises:
 30            QQError if the string corresponds to no transfer mode.
 31        """
 32        match s.lower().strip():
 33            case "always":
 34                return Always()
 35            case "never":
 36                return Never()
 37            case "success":
 38                return Success()
 39            case "failure":
 40                return Failure()
 41            case _:
 42                # if the string is a number (positive or negative)
 43                if bool(re.match(r"^-?\d+$", s.strip())):
 44                    return ExitCode(int(s))
 45
 46                raise QQError(f"Could not recognize a transfer mode variant '{s}'.")
 47
 48    @classmethod
 49    def multi_from_str(cls, raw: str) -> list["TransferMode"]:
 50        """
 51        Parse a string containing multiple transfer modes.
 52
 53        Args:
 54            s: String containing transfer mode variants separated by a colon, comma, or space.
 55                Example: "success:42", "1,2,3", or "failure success".
 56
 57        Returns:
 58            list[TransferMode]: A list of parsed transfer modes.
 59
 60        Raises:
 61            QQError: If any of the individual mode strings cannot be recognized.
 62        """
 63        mode_strings = re.split(r"[:,\s]+", raw.strip())
 64        mode_strings = [ms for ms in mode_strings if ms]
 65
 66        return [TransferMode.from_str(mode_str) for mode_str in mode_strings]
 67
 68    @abstractmethod
 69    def should_transfer(self, exit_code: int) -> bool:
 70        """
 71        Determine whether data should be transferred/archived based on the exit code.
 72
 73        Args:
 74            exit_code: The exit code of the completed job.
 75
 76        Returns:
 77            True if data should be transferred/archived, False otherwise.
 78        """
 79
 80    @abstractmethod
 81    def to_str(self) -> str:
 82        """
 83        Convert the TransferMode variant to its string representation.
 84
 85        Returns:
 86            A string representation of the transfer mode variant.
 87        """
 88
 89
 90@dataclass(frozen=True)
 91class Always(TransferMode):
 92    """
 93    Data are always transferred/archived regardless the job's exit code.
 94    """
 95
 96    def should_transfer(self, exit_code: int) -> bool:
 97        _ = exit_code
 98        return True
 99
100    def to_str(self) -> str:
101        return "always"
102
103
104@dataclass(frozen=True)
105class Never(TransferMode):
106    """
107    Data are never transferred/archived.
108    """
109
110    def should_transfer(self, exit_code: int) -> bool:
111        _ = exit_code
112        return False
113
114    def to_str(self) -> str:
115        return "never"
116
117
118@dataclass(frozen=True)
119class Success(TransferMode):
120    """
121    Data are transferred/archived only if the job completes successfully.
122    """
123
124    def should_transfer(self, exit_code: int) -> bool:
125        return exit_code == 0
126
127    def to_str(self) -> str:
128        return "success"
129
130
131@dataclass(frozen=True)
132class Failure(TransferMode):
133    """
134    Data are transferred/archived only if the job fails.
135    """
136
137    def should_transfer(self, exit_code: int) -> bool:
138        return exit_code != 0
139
140    def to_str(self) -> str:
141        return "failure"
142
143
144@dataclass(frozen=True)
145class ExitCode(TransferMode):
146    """
147    Data are transferred/archived only if the job exits with a specific code.
148
149    Attributes:
150        code: The exit code that triggers data transfer.
151    """
152
153    code: int
154
155    def should_transfer(self, exit_code: int) -> bool:
156        return exit_code == self.code
157
158    def to_str(self) -> str:
159        return f"{self.code}"