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

0.0

/cg/services/post_processing/pacbio/post_processing_service.py

from cg.services.post_processing.abstract_classes import PostProcessingService
from cg.services.post_processing.error_handlers import handle_post_processing_errors
from cg.services.post_processing.exc import (
    PostProcessingStoreDataError,
    PostProcessingError,
    PostProcessingStoreFileError,
    PostProcessingRunDataGeneratorError,
)
from cg.services.post_processing.pacbio.data_storage_service.pacbio_store_service import (
    PacBioStoreService,
)
from cg.services.post_processing.pacbio.housekeeper_service.pacbio_houskeeper_service import (
    PacBioHousekeeperService,
)
from cg.services.post_processing.pacbio.run_data_generator.pacbio_run_data_generator import (
    PacBioRunDataGenerator,
)
from cg.services.post_processing.pacbio.run_data_generator.run_data import PacBioRunData


class PacBioPostProcessingService(PostProcessingService):
    """Service for handling post-processing of PacBio sequencing runs."""

    def __init__(
        self,
        run_data_generator: PacBioRunDataGenerator,
        hk_service: PacBioHousekeeperService,
        store_service: PacBioStoreService,
    ):
        self.run_data_generator: PacBioRunDataGenerator = run_data_generator
        self.hk_service: PacBioHousekeeperService = hk_service
        self.store_service: PacBioStoreService = store_service

    @handle_post_processing_errors(
        to_except=(
            PostProcessingStoreDataError,
            PostProcessingRunDataGeneratorError,
            PostProcessingStoreFileError,
        ),
        to_raise=PostProcessingError,
    )
    def post_process(self, run_name: str, sequencing_dir: str):
        run_data: PacBioRunData = self.run_data_generator.get_run_data(
            run_name=run_name, sequencing_dir=sequencing_dir
        )
        self.store_service.store_post_processing_data(run_data=run_data)
        self.hk_service.store_files_in_housekeeper(run_data=run_data)

1	from cg.services.post_processing.abstract_classes import PostProcessingService	×
NEW 2	from cg.services.post_processing.error_handlers import handle_post_processing_errors	×
NEW 3	from cg.services.post_processing.exc import (	×
4	PostProcessingStoreDataError,
5	PostProcessingError,
6	PostProcessingStoreFileError,
7	PostProcessingRunDataGeneratorError,
8	)
9	from cg.services.post_processing.pacbio.data_storage_service.pacbio_store_service import (	×
10	PacBioStoreService,
11	)
12	from cg.services.post_processing.pacbio.housekeeper_service.pacbio_houskeeper_service import (	×
13	PacBioHousekeeperService,
14	)
15	from cg.services.post_processing.pacbio.run_data_generator.pacbio_run_data_generator import (	×
16	PacBioRunDataGenerator,
17	)
18	from cg.services.post_processing.pacbio.run_data_generator.run_data import PacBioRunData	×
19
20
21	class PacBioPostProcessingService(PostProcessingService):	×
22	"""Service for handling post-processing of PacBio sequencing runs."""
23
24	def __init__(	×
25	self,
26	run_data_generator: PacBioRunDataGenerator,
27	hk_service: PacBioHousekeeperService,
28	store_service: PacBioStoreService,
29	):
30	self.run_data_generator: PacBioRunDataGenerator = run_data_generator	×
31	self.hk_service: PacBioHousekeeperService = hk_service	×
32	self.store_service: PacBioStoreService = store_service	×
33
NEW 34	@handle_post_processing_errors(	×
35	to_except=(
36	PostProcessingStoreDataError,
37	PostProcessingRunDataGeneratorError,
38	PostProcessingStoreFileError,
39	),
40	to_raise=PostProcessingError,
41	)
42	def post_process(self, run_name: str, sequencing_dir: str):	×
43	run_data: PacBioRunData = self.run_data_generator.get_run_data(	×
44	run_name=run_name, sequencing_dir=sequencing_dir
45	)
46	self.store_service.store_post_processing_data(run_data=run_data)	×
47	self.hk_service.store_files_in_housekeeper(run_data=run_data)	×

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