10162085953

Committed 30 Jul 2024 12:12PM UTC coverage: 84.403%. First build

Build # 10162085953

Build Type

Pull #3487

github

Committed by

web-flow

Commit Message

Merge a56ac7a6b into 8380cf2de

Pull Request Pull Request #3487: add error handlers pacbio flow

Run Details

46 of 68 new or added lines in 11 files covered. (67.65%)

21121 of 25024 relevant lines covered (84.4%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.47

/cg/services/post_processing/abstract_classes.py

"""Post-processing service abstract classes."""

from abc import ABC, abstractmethod
from pathlib import Path

from cg.services.post_processing.abstract_models import PostProcessingDTOs, RunData, RunMetrics


class RunDataGenerator(ABC):
    """Abstract class for that holds functionality to create a run data model."""

    @abstractmethod
    def get_run_data(self, run_name: str, sequencing_dir: str) -> RunData:
        raise NotImplementedError


class RunFileManager(ABC):
    """Abstract class that manages files related to an instrument run."""

    @abstractmethod
    def get_files_to_parse(self, run_data: RunData) -> list[Path]:
        """Get the files required for the PostProcessingMetricsService."""
        raise NotImplementedError

    @abstractmethod
    def get_files_to_store(self, run_data: RunData) -> list[Path]:
        """Get the files to store for the PostProcessingHKService."""
        raise NotImplementedError


class PostProcessingMetricsParser(ABC):
    """Abstract class that manages the metrics parsing related to an instrument run."""

    @abstractmethod
    def parse_metrics(self, run_data: RunData) -> RunMetrics:
        raise NotImplementedError


class PostProcessingDataTransferService(ABC):
    """Abstract class that manages the data transfer from parsed metrics to the database structure."""

    @abstractmethod
    def get_post_processing_dtos(self, run_data: RunData) -> PostProcessingDTOs:
        raise NotImplementedError


class PostProcessingStoreService(ABC):
    """Abstract class that manages storing data transfer objects in the database."""

    @abstractmethod
    def store_post_processing_data(self, run_data: RunData):
        raise NotImplementedError


class PostProcessingHKService(ABC):
    """Abstract class that manages storing of files for an instrument run."""

    @abstractmethod
    def store_files_in_housekeeper(self, run_data: RunData):
        raise NotImplementedError


class PostProcessingService(ABC):
    """Abstract class that encapsulates the logic required for post-processing and instrument run."""

    @abstractmethod
    def post_process(self, run_name: str, sequencing_dir: str):
        """Store sequencing metrics in statusdb and relevant files in housekeeper"""
        raise NotImplementedError

1	"""Post-processing service abstract classes."""
2
3	from abc import ABC, abstractmethod	1✔
4	from pathlib import Path	1✔
5
6	from cg.services.post_processing.abstract_models import PostProcessingDTOs, RunData, RunMetrics	1✔
7
8
9	class RunDataGenerator(ABC):	1✔
10	"""Abstract class for that holds functionality to create a run data model."""
11
12	@abstractmethod	1✔
13	def get_run_data(self, run_name: str, sequencing_dir: str) -> RunData:	1✔
NEW 14	raise NotImplementedError	×
15
16
17	class RunFileManager(ABC):	1✔
18	"""Abstract class that manages files related to an instrument run."""
19
20	@abstractmethod	1✔
21	def get_files_to_parse(self, run_data: RunData) -> list[Path]:	1✔
22	"""Get the files required for the PostProcessingMetricsService."""
NEW 23	raise NotImplementedError	×
24
25	@abstractmethod	1✔
26	def get_files_to_store(self, run_data: RunData) -> list[Path]:	1✔
27	"""Get the files to store for the PostProcessingHKService."""
NEW 28	raise NotImplementedError	×
29
30
31	class PostProcessingMetricsParser(ABC):	1✔
32	"""Abstract class that manages the metrics parsing related to an instrument run."""
33
34	@abstractmethod	1✔
35	def parse_metrics(self, run_data: RunData) -> RunMetrics:	1✔
NEW 36	raise NotImplementedError	×
37
38
39	class PostProcessingDataTransferService(ABC):	1✔
40	"""Abstract class that manages the data transfer from parsed metrics to the database structure."""
41
42	@abstractmethod	1✔
43	def get_post_processing_dtos(self, run_data: RunData) -> PostProcessingDTOs:	1✔
NEW 44	raise NotImplementedError	×
45
46
47	class PostProcessingStoreService(ABC):	1✔
48	"""Abstract class that manages storing data transfer objects in the database."""
49
50	@abstractmethod	1✔
51	def store_post_processing_data(self, run_data: RunData):	1✔
NEW 52	raise NotImplementedError	×
53
54
55	class PostProcessingHKService(ABC):	1✔
56	"""Abstract class that manages storing of files for an instrument run."""
57
58	@abstractmethod	1✔
59	def store_files_in_housekeeper(self, run_data: RunData):	1✔
NEW 60	raise NotImplementedError	×
61
62
63	class PostProcessingService(ABC):	1✔
64	"""Abstract class that encapsulates the logic required for post-processing and instrument run."""
65
66	@abstractmethod	1✔
67	def post_process(self, run_name: str, sequencing_dir: str):	1✔
68	"""Store sequencing metrics in statusdb and relevant files in housekeeper"""
NEW 69	raise NotImplementedError	×

Clinical-Genomics / cg / 10162085953

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous