Improved after feedback

Author: RasmusBurge-CG

cg/services/illumina/backup/backup_service.py

@@ -1,4 +1,5 @@
 import subprocess
+from datetime import datetime
 from pathlib import Path
 
 from cg.apps.slurm.slurm_api import SlurmAPI
@@ -11,24 +12,21 @@
     IlluminaRunEncryptionError,
     PdcError,
     PdcNoFilesMatchingSearchError,
-    ValidationError,
 )
 from cg.meta.backup.backup import LOG
 from cg.meta.encryption.encryption import EncryptionAPI
 from cg.meta.tar.tar import TarAPI
 from cg.models.cg_config import PDCArchivingDirectory
 from cg.models.run_devices.illumina_run_directory_data import IlluminaRunDirectoryData
 from cg.services.illumina.backup.encrypt_service import IlluminaRunEncryptionService
+from cg.services.illumina.backup.models import DsmcEncryptionKey, DsmcSequencingFile
 from cg.services.illumina.backup.utils import (
     DsmcOutput,
+    convert_string_to_datetime_object,
     get_latest_dsmc_archived_sequencing_run,
     get_latest_dsmc_encryption_key,
-    contains_dsmc_key,
-    contains_dsmc_sequencing_path,
-)
-from cg.services.illumina.file_parsing.models import (
-    DsmcEncryptionKey,
-    DsmcSequencingFile,
+    is_dsmc_encryption_key,
+    is_dsmc_sequencing_path,
 )
 from cg.services.pdc_service.pdc_service import PdcService
 from cg.store.models import IlluminaSequencingRun
@@ -293,58 +291,71 @@ def parse_dsmc_output_sequencing_path(dsmc_output: list[str]) -> list[DsmcSequen
         """Parses the DSMC command output to extract validated sequencing paths."""
         validated_responses = []
         for line in dsmc_output:
-            if contains_dsmc_sequencing_path(line):
+            if is_dsmc_sequencing_path(line):
                 parts = line.split()
-                try:
-                    query_response = DsmcSequencingFile(
-                        date=f"{parts[DsmcOutput.DATE_COLUMN_INDEX]} {parts[DsmcOutput.TIME_COLUMN_INDEX]}",
-                        sequencing_path=parts[DsmcOutput.PATH_COLUMN_INDEX],
-                    )
-                    validated_responses.append(query_response)
-                except ValidationError as e:
-                    LOG.error(f"Validation error for line: {line}\nError: {e}")
+
+                fileDateTime: datetime = convert_string_to_datetime_object(
+                    f"{parts[DsmcOutput.DATE_COLUMN_INDEX]} {parts[DsmcOutput.TIME_COLUMN_INDEX]}"
+                )
+
+                query_response = DsmcSequencingFile(
+                    dateTime=fileDateTime,
+                    path=Path(parts[DsmcOutput.PATH_COLUMN_INDEX]),
+                )
+                validated_responses.append(query_response)
 
         return validated_responses
 
     @classmethod
     def get_latest_archived_sequencing_run_path(cls, dsmc_output: list[str]) -> Path | None:
         """Get the path of the archived sequencing run from a PDC query."""
-        validated_sequencing_paths = cls.parse_dsmc_output_sequencing_path(dsmc_output)
+        validated_sequencing_paths: list[DsmcSequencingFile] = (
+            cls.parse_dsmc_output_sequencing_path(dsmc_output)
+        )
 
-        archived_run = get_latest_dsmc_archived_sequencing_run(validated_sequencing_paths)
+        archived_run: DsmcSequencingFile = get_latest_dsmc_archived_sequencing_run(
+            validated_sequencing_paths
+        )
 
         if archived_run:
             LOG.info(f"Sequencing run found: {archived_run}")
-            return archived_run
+            return archived_run.path
 
     @staticmethod
     def parse_dsmc_output_key_path(dsmc_output: list[str]) -> list[DsmcEncryptionKey]:
         """Parses the DSMC command output to extract validated encryption keys."""
         validated_responses = []
         for line in dsmc_output:
-            if contains_dsmc_key(line):
-                parts = line.split()
-                try:
-                    query_response = DsmcEncryptionKey(
-                        date=f"{parts[DsmcOutput.DATE_COLUMN_INDEX]} {parts[DsmcOutput.TIME_COLUMN_INDEX]}",
-                        key_path=parts[DsmcOutput.PATH_COLUMN_INDEX],
-                    )
-                    validated_responses.append(query_response)
-                except ValidationError as e:
-                    LOG.error(f"Validation error for line: {line}\nError: {e}")
+            if is_dsmc_encryption_key(line):
+                parts: list[str] = line.split()
+
+                fileDateTime: datetime = convert_string_to_datetime_object(
+                    f"{parts[DsmcOutput.DATE_COLUMN_INDEX]} {parts[DsmcOutput.TIME_COLUMN_INDEX]}"
+                )
+
+                query_response = DsmcEncryptionKey(
+                    dateTime=fileDateTime,
+                    path=Path(parts[DsmcOutput.PATH_COLUMN_INDEX]),
+                )
+
+                validated_responses.append(query_response)
 
         return validated_responses
 
     @classmethod
     def get_archived_encryption_key_path(cls, dsmc_output: list[str]) -> Path | None:
         """Get the encryption key for the archived sequencing run from a PDC query."""
-        validated_encryption_keys = cls.parse_dsmc_output_key_path(dsmc_output)
+        validated_encryption_keys: list[DsmcEncryptionKey] = cls.parse_dsmc_output_key_path(
+            dsmc_output
+        )
 
-        archived_encryption_key = get_latest_dsmc_encryption_key(validated_encryption_keys)
+        archived_encryption_key: DsmcEncryptionKey = get_latest_dsmc_encryption_key(
+            validated_encryption_keys
+        )
 
         if archived_encryption_key:
             LOG.info(f"Encryption key found: {archived_encryption_key}")
-            return archived_encryption_key
+            return archived_encryption_key.path
 
     def validate_is_run_backup_possible(
         self,

cg/services/illumina/backup/exc.py

@@ -0,0 +1,9 @@
+from cg.exc import CgError
+
+
+class DsmcMissingSequenceFileError(CgError):
+    """Exception raised when a Dsmc sequence file is not found."""
+
+
+class DsmcMissingEncryptionKeyError(CgError):
+    """Exception raised when a Dsmc encryption key is not found."""

cg/services/illumina/backup/models.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+from pathlib import Path
+from typing import Annotated
+
+from pydantic import BaseModel, BeforeValidator
+
+from cg.services.illumina.backup.validators import (
+    is_valid_dsmc_encryption_key_path,
+    is_valid_dsmc_sequencing_file_path,
+)
+
+
+class DsmcEncryptionKey(BaseModel):
+    """Model representing the response from a PDC query."""
+
+    dateTime: datetime
+    path: Annotated[Path, BeforeValidator(is_valid_dsmc_encryption_key_path)]
+
+
+class DsmcSequencingFile(BaseModel):
+    """Model representing the response from a PDC query."""
+
+    dateTime: datetime
+    path: Annotated[Path, BeforeValidator(is_valid_dsmc_sequencing_file_path)]

cg/services/illumina/backup/utils.py

@@ -1,15 +1,14 @@
 """Helper functions."""
 
-from enum import IntEnum
-from pathlib import Path
+from datetime import datetime
+from operator import attrgetter
 
 from cg.constants import FileExtensions
-from cg.exc import ValidationError
-from cg.services.illumina.file_parsing.models import (
-    DsmcEncryptionKey,
-    DsmcSequencingFile,
+from cg.services.illumina.backup.exc import (
+    DsmcMissingEncryptionKeyError,
+    DsmcMissingSequenceFileError,
 )
-from cg.constants import FileExtensions
+from cg.services.illumina.backup.models import DsmcEncryptionKey, DsmcSequencingFile
 
 
 class DsmcOutput:
@@ -18,41 +17,54 @@ class DsmcOutput:
     PATH_COLUMN_INDEX = 4
 
 
-def contains_dsmc_key(line: str) -> bool:
-    if (
+def is_dsmc_encryption_key(line: str) -> bool:
+    return (
         FileExtensions.KEY in line
         and FileExtensions.GPG in line
         and FileExtensions.GZIP not in line
-    ):
-        return True
-    return False
+    )
 
 
-def contains_dsmc_sequencing_path(line: str) -> bool:
-    if FileExtensions.TAR in line and FileExtensions.GZIP in line and FileExtensions.GPG in line:
-        return True
-    return False
+def is_dsmc_sequencing_path(line: str) -> bool:
+    return FileExtensions.TAR in line and FileExtensions.GZIP in line and FileExtensions.GPG in line
 
 
-def get_latest_dsmc_archived_sequencing_run(dsmc_files: list[DsmcSequencingFile]) -> Path:
+def get_latest_dsmc_archived_sequencing_run(
+    dsmc_files: list[DsmcSequencingFile],
+) -> DsmcSequencingFile:
     """Return the latest file path based on the date attribute."""
+
     if not dsmc_files:
-        return None  # Return None if the list is empty
+        raise DsmcMissingSequenceFileError("No archived sequencing in DSMC output.")
 
-    # Get the file with the latest date
-    latest_file = max(dsmc_files, key=lambda file: file.date)
+    latest_file = max(dsmc_files, key=attrgetter("dateTime"))
 
-    # Return the sequencing_path as a Path object
-    return Path(latest_file.sequencing_path)
+    return latest_file
 
 
-def get_latest_dsmc_encryption_key(dsmc_files: list[DsmcEncryptionKey]) -> Path:
+def get_latest_dsmc_encryption_key(dsmc_files: list[DsmcEncryptionKey]) -> DsmcEncryptionKey:
     """Return the latest file path based on the date attribute."""
+
     if not dsmc_files:
-        return None  # Return None if the list is empty
+        raise DsmcMissingEncryptionKeyError("No Encryption Key in DSMC output.")
+
+    latest_file = max(dsmc_files, key=attrgetter("dateTime"))
+
+    return latest_file
+
+
+def convert_string_to_datetime_object(strDateTime: str) -> datetime:
+    date_formats = [
+        "%Y-%m-%d %H:%M:%S",
+        "%Y/%m/%d %H:%M:%S",
+        "%d/%m/%Y %H:%M:%S",
+        "%m/%d/%Y %H:%M:%S",
+    ]
 
-    # Get the file with the latest date
-    latest_file = max(dsmc_files, key=lambda file: file.date)
+    for fmt in date_formats:
+        try:
+            return datetime.strptime(strDateTime, fmt)
+        except ValueError:
+            continue
 
-    # Return the sequencing_path as a Path object
-    return Path(latest_file.key_path)
+    raise ValueError(f"Could not convert '{strDateTime}' to a datetime object.")

cg/services/illumina/backup/validators.py

@@ -0,0 +1,16 @@
+from pathlib import Path
+
+from cg.constants import FileExtensions
+from cg.exc import MissingFilesError
+
+
+def is_valid_dsmc_encryption_key_path(value: Path) -> Path:
+    if not value.name.endswith(f"{FileExtensions.KEY}{FileExtensions.GPG}"):
+        raise MissingFilesError("Missing a valid encryption key.")
+    return value
+
+
+def is_valid_dsmc_sequencing_file_path(value: Path) -> Path:
+    if not value.name.endswith(f"{FileExtensions.TAR}{FileExtensions.GZIP}{FileExtensions.GPG}"):
+        raise MissingFilesError("Missing a valid sequence file.")
+    return value

cg/services/illumina/file_parsing/models.py

@@ -1,12 +1,6 @@
-from datetime import datetime
-from pathlib import Path
+from pydantic import BaseModel, Field
 
-from pydantic import BaseModel, Field, field_validator
-
-from cg.constants import FileExtensions, SequencingRunDataAvailability
-from cg.constants.devices import DeviceType
 from cg.constants.metrics import DemuxMetricsColumnNames, QualityMetricsColumnNames
-from cg.constants.sequencing import Sequencers
 
 
 class SequencingQualityMetrics(BaseModel):
@@ -29,41 +23,3 @@ class DemuxMetrics(BaseModel):
     lane: int = Field(..., alias=DemuxMetricsColumnNames.LANE)
     sample_internal_id: str = Field(..., alias=DemuxMetricsColumnNames.SAMPLE_INTERNAL_ID)
     read_pair_count: int = Field(..., alias=DemuxMetricsColumnNames.READ_PAIR_COUNT)
-
-
-class DsmcEncryptionKey(BaseModel):
-    """Model representing the response from a PDC query."""
-
-    date: str
-    key_path: str
-
-    @field_validator("date")
-    def parse_date(cls, value: str) -> datetime:
-        return datetime.strptime(value, "%m/%d/%Y %H:%M:%S")
-
-    @field_validator("key_path")
-    def validate_sequencing_path(cls, value: str) -> str:
-        if not value.endswith(f"{FileExtensions.KEY}{FileExtensions.GPG}"):
-            raise ValueError(f'"{value}" - is not the path to the Encryption key')
-        if not Path(value):
-            raise ValueError(f'"{value}" - is not a valid file path.')
-        return value
-
-
-class DsmcSequencingFile(BaseModel):
-    """Model representing the response from a PDC query."""
-
-    date: str
-    sequencing_path: str
-
-    @field_validator("date")
-    def parse_date(cls, value: str) -> datetime:
-        return datetime.strptime(value, "%m/%d/%Y %H:%M:%S")
-
-    @field_validator("sequencing_path")
-    def validate_sequencing_path(cls, value: str) -> str:
-        if not value.endswith(f"{FileExtensions.TAR}{FileExtensions.GZIP}{FileExtensions.GPG}"):
-            raise ValueError(f'"{value}" - is not the path to the archived sequencing file')
-        if not Path(value):
-            raise ValueError(f'"{value}" - is not a valid file path.')
-        return value

tests/services/illumina/backup/test_backup_services.py

@@ -2,6 +2,7 @@
 
 import fnmatch
 import logging
+from datetime import datetime
 from pathlib import Path
 from typing import Callable
 
@@ -18,6 +19,7 @@
 from cg.models.cg_config import CGConfig, PDCArchivingDirectory
 from cg.services.illumina.backup.backup_service import IlluminaBackupService
 from cg.services.illumina.backup.encrypt_service import IlluminaRunEncryptionService
+from cg.services.illumina.backup.utils import convert_string_to_datetime_object
 from cg.services.pdc_service.pdc_service import PdcService
 from cg.store.models import IlluminaSequencingRun
 from cg.store.store import Store
@@ -137,6 +139,40 @@ def test_maximum_processing_queue_full(store_with_illumina_sequencing_data: Stor
     assert backup_api.has_processing_queue_capacity() is False
 
 
+def test_convert_string_to_datetime_object_valid():
+    # GIVEN a list of valid datetime strings with different formats
+    valid_datetime_strings = [
+        ("2024-10-15 12:45:30", "%Y-%m-%d %H:%M:%S"),
+        ("2024/10/15 12:45:30", "%Y/%m/%d %H:%M:%S"),
+        ("15/10/2024 12:45:30", "%d/%m/%Y %H:%M:%S"),
+        ("10/15/2024 12:45:30", "%m/%d/%Y %H:%M:%S"),
+    ]
+
+    # WHEN the function is called with valid strings
+    for date_str, expected_format in valid_datetime_strings:
+        # THEN it should return the correct datetime object
+        expected_datetime = datetime.strptime(date_str, expected_format)
+        assert convert_string_to_datetime_object(date_str) == expected_datetime
+
+
+def test_convert_string_to_datetime_object_invalid():
+    # GIVEN a list of invalid datetime strings
+    invalid_datetime_strings = [
+        "2024-15-10 12:45:30",  # Invalid day format
+        "15-10-2024 12:45",  # Missing seconds
+        "Invalid string",  # Completely invalid
+    ]
+
+    # WHEN the function is called with invalid strings
+    for date_str in invalid_datetime_strings:
+        # THEN it should raise a ValueError
+        try:
+            convert_string_to_datetime_object(date_str)
+            assert False, f"Expected ValueError for {date_str} but it didn't raise"
+        except ValueError as e:
+            assert str(e) == f"Could not convert '{date_str}' to a datetime object."
+
+
 def test_maximum_processing_queue_not_full(store_with_illumina_sequencing_data: Store):
     # GIVEN a store with a requested sequencing run
     sequencing_runs: list[IlluminaSequencingRun] = store_with_illumina_sequencing_data._get_query(