8892829037

Committed 30 Apr 2024 09:55AM UTC coverage: 84.455%. First build

Build # 8892829037

Build Type

Pull #3148

github

Committed by

web-flow

Commit Message

Merge 8a772805d into 73c972545

Pull Request Pull Request #3148: add(create validation case)

Run Details

153 of 188 new or added lines in 6 files covered. (81.38%)

19776 of 23416 relevant lines covered (84.46%)

0.84 hits per line

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

92.63

/cg/meta/create_validation_cases/validation_cases_api.py

"""Module that holds the api to create validation cases."""

import logging
import os
from pathlib import Path

from housekeeper.store.models import File, Version

from cg.apps.housekeeper.hk import HousekeeperAPI
from cg.constants import SequencingFileTag
from cg.constants.housekeeper_tags import VALIDATION_TAG
from cg.meta.create_validation_cases.validation_case_data import ValidationCaseData
from cg.meta.create_validation_cases.validation_data_input import ValidationDataInput
from cg.store.models import Case, CaseSample, Sample
from cg.store.store import Store
from cg.utils.files import copy_file, get_files_matching_pattern, rename_file

LOG = logging.getLogger(__name__)


class CreateValidationCaseAPI:

    def __init__(self, status_db: Store, housekeeper_api: HousekeeperAPI, dry_run: bool = False):
        self.status_db: Store = status_db
        self.hk_api: HousekeeperAPI = housekeeper_api
        self.dry_run = dry_run

    def get_validation_case_data(self, validation_data_input: ValidationDataInput):
        return ValidationCaseData(
            status_db=self.status_db,
            validation_data_input=validation_data_input,
        )

    def store_validation_samples(self, validation_case_data: ValidationCaseData) -> None:
        """
        Add a validation sample entry to StatusDB.
        Raises:
            ValueError
        """
        for sample in validation_case_data.validation_samples:
            if self.status_db.sample_with_id_exists(sample_id=sample.internal_id):
                raise ValueError(f"Sample {sample.internal_id} already exists in StatusDB.")
            LOG.info(
                f"New validation sample created: {sample.internal_id} from {sample.from_sample}"
                f"Application tag set to: {sample.application_version.application.tag}"
            )
            if not self.dry_run:
                self.status_db.session.add(sample)
                LOG.info(f"Added {sample.name} to StatusDB.")

    def store_validation_case(self, validation_case_data: ValidationCaseData) -> None:
        """
        Add a validation case entry to StatusDB.
        """
        validation_case: Case = validation_case_data.validation_case
        if self.status_db.case_with_name_exists(case_name=validation_case.name):
            LOG.info(f"Case with name {validation_case.name} already exists in StatusDB.")
            return
        if not self.dry_run:
            self.status_db.session.add(validation_case)
            LOG.info(f"New validation case created: {validation_case.internal_id}")

    def _link_sample_to_case(self, validation_case_data: ValidationCaseData) -> None:
        """Create a link between sample and case in statusDB."""
        for sample in validation_case_data.validation_samples:
            sample_case_link: CaseSample = self.status_db.relate_sample(
                case=validation_case_data.validation_case,
                sample=sample,
                status=self._get_sample_status(sample),
            )
            if self.dry_run:
                return
            self.status_db.session.add(sample_case_link)
            LOG.info(
                f"Related sample {sample.internal_id} to {validation_case_data.validation_case.internal_id}"
            )

    @staticmethod
    def _get_sample_status(sample: Sample) -> str:
        """Return the status of a sample."""
        return sample.links[0].status if sample.links else "unknown"

    def create_validation_case_in_statusdb(self, validation_case_data: ValidationCaseData) -> None:
        """
        Add the validation samples and case to statusDB and generate the sample case link.
        """
        try:
            self.store_validation_samples(validation_case_data)
            self.store_validation_case(validation_case_data)
            self._link_sample_to_case(validation_case_data)
            self.status_db.session.commit()
        except Exception as error:
            LOG.warning(f"An error occurred {repr(error)}")
            self.status_db.session.rollback()

    def create_validation_sample_bundles(self, validation_samples: list[Sample]) -> None:
        for validation_sample in validation_samples:
            self.hk_api.create_new_bundle_and_version(validation_sample.internal_id)

    def copy_original_sample_bundle_files(self, validation_samples: list[Sample]) -> None:
        """Copy the fastq files from the original sample bundle into the new validation sample bundle"""
        for validation_sample in validation_samples:
            fastq_files: list[File] = self.hk_api.get_files_from_latest_version(
                bundle_name=validation_sample.from_sample, tags=[SequencingFileTag.FASTQ]
            )
            validation_bundle_path: Path = self.hk_api.get_latest_bundle_version(
                bundle_name=validation_sample.internal_id
            ).full_path
            if not validation_bundle_path.exists():
                os.makedirs(validation_bundle_path)
            for fastq_file in fastq_files:
                copy_file(file_path=Path(fastq_file.full_path), destination=validation_bundle_path)

    def rename_and_add_original_sample_bundle_files(self, validation_samples: list[Sample]) -> None:
        """Rename the fastq files to use the new validation sample internal id."""
        for validation_sample in validation_samples:
            validation_bundle_path: Path = self.hk_api.get_latest_bundle_version(
                bundle_name=validation_sample.internal_id
            ).full_path
            fastq_files: list[Path] = get_files_matching_pattern(
                directory=validation_bundle_path, pattern="*." + SequencingFileTag.FASTQ + ".*"
            )
            for fastq_file in fastq_files:
                new_fast_file_path = self.get_new_fastq_file_path(
                    fastq_file=fastq_file, validation_sample=validation_sample
                )
                rename_file(file_path=fastq_file, renamed_file_path=new_fast_file_path)
                new_tags: list[str] = self.get_new_tags(validation_sample)
                self.hk_api.add_and_include_file_to_latest_version(
                    bundle_name=validation_sample.internal_id,
                    file=new_fast_file_path,
                    tags=new_tags,
                )

    def get_new_tags(self, validation_sample: Sample) -> list[str]:
        """
        Get the fastq file tags from the original sample and
        replace the bundle name with the validation sample internal id.
        """
        hk_files: list[File] = self.hk_api.get_files_from_latest_version(
            bundle_name=validation_sample.from_sample, tags=[SequencingFileTag.FASTQ]
        )
        original_tags: list[str] = self.hk_api.get_tag_names_from_file(hk_files[0])
        new_tags: list[str] = [validation_sample.internal_id, VALIDATION_TAG]
        for tag in original_tags:
            if tag != validation_sample.from_sample:
                new_tags.append(tag)
        return new_tags

    @staticmethod
    def get_new_fastq_file_path(fastq_file: Path, validation_sample: Sample) -> Path:
        return Path(
            fastq_file.as_posix().replace(
                validation_sample.from_sample,
                validation_sample.internal_id,
            )
        )

    def create_validation_samples_in_housekeeper(self, validation_case_data: ValidationCaseData):
        self.create_validation_sample_bundles(validation_case_data.validation_samples)
        self.copy_original_sample_bundle_files(validation_case_data.validation_samples)
        self.rename_and_add_original_sample_bundle_files(validation_case_data.validation_samples)

    def create_validation_case(
        self,
        case_id: str,
        case_name: str,
        data_analysis: str | None = None,
        delivery: str | None = None,
    ) -> None:
        """Create the validation case in statusdb and associated sample bundles in housekeeper."""
        validation_data_input = ValidationDataInput(
            case_id=case_id, case_name=case_name, data_analysis=data_analysis, delivery=delivery
        )
        validation_case_data: ValidationCaseData = self.get_validation_case_data(
            validation_data_input
        )
        self.create_validation_case_in_statusdb(validation_case_data)
        self.create_validation_samples_in_housekeeper(validation_case_data)

1	"""Module that holds the api to create validation cases."""
2
3	import logging	1✔
4	import os	1✔
5	from pathlib import Path	1✔
6
7	from housekeeper.store.models import File, Version	1✔
8
9	from cg.apps.housekeeper.hk import HousekeeperAPI	1✔
10	from cg.constants import SequencingFileTag	1✔
11	from cg.constants.housekeeper_tags import VALIDATION_TAG	1✔
12	from cg.meta.create_validation_cases.validation_case_data import ValidationCaseData	1✔
13	from cg.meta.create_validation_cases.validation_data_input import ValidationDataInput	1✔
14	from cg.store.models import Case, CaseSample, Sample	1✔
15	from cg.store.store import Store	1✔
16	from cg.utils.files import copy_file, get_files_matching_pattern, rename_file	1✔
17
18	LOG = logging.getLogger(__name__)	1✔
19
20
21	class CreateValidationCaseAPI:	1✔
22
23	def __init__(self, status_db: Store, housekeeper_api: HousekeeperAPI, dry_run: bool = False):	1✔
24	self.status_db: Store = status_db	1✔
25	self.hk_api: HousekeeperAPI = housekeeper_api	1✔
26	self.dry_run = dry_run	1✔
27
28	def get_validation_case_data(self, validation_data_input: ValidationDataInput):	1✔
29	return ValidationCaseData(	1✔
30	status_db=self.status_db,
31	validation_data_input=validation_data_input,
32	)
33
34	def store_validation_samples(self, validation_case_data: ValidationCaseData) -> None:	1✔
35	"""
36	Add a validation sample entry to StatusDB.
37	Raises:
38	ValueError
39	"""
40	for sample in validation_case_data.validation_samples:	1✔
41	if self.status_db.sample_with_id_exists(sample_id=sample.internal_id):	1✔
NEW 42	raise ValueError(f"Sample {sample.internal_id} already exists in StatusDB.")	×
43	LOG.info(	1✔
44	f"New validation sample created: {sample.internal_id} from {sample.from_sample}"
45	f"Application tag set to: {sample.application_version.application.tag}"
46	)
47	if not self.dry_run:	1✔
48	self.status_db.session.add(sample)	1✔
49	LOG.info(f"Added {sample.name} to StatusDB.")	1✔
50
51	def store_validation_case(self, validation_case_data: ValidationCaseData) -> None:	1✔
52	"""
53	Add a validation case entry to StatusDB.
54	"""
55	validation_case: Case = validation_case_data.validation_case	1✔
56	if self.status_db.case_with_name_exists(case_name=validation_case.name):	1✔
NEW 57	LOG.info(f"Case with name {validation_case.name} already exists in StatusDB.")	×
NEW 58	return	×
59	if not self.dry_run:	1✔
60	self.status_db.session.add(validation_case)	1✔
61	LOG.info(f"New validation case created: {validation_case.internal_id}")	1✔
62
63	def _link_sample_to_case(self, validation_case_data: ValidationCaseData) -> None:	1✔
64	"""Create a link between sample and case in statusDB."""
65	for sample in validation_case_data.validation_samples:	1✔
66	sample_case_link: CaseSample = self.status_db.relate_sample(	1✔
67	case=validation_case_data.validation_case,
68	sample=sample,
69	status=self._get_sample_status(sample),
70	)
71	if self.dry_run:	1✔
NEW 72	return	×
73	self.status_db.session.add(sample_case_link)	1✔
74	LOG.info(	1✔
75	f"Related sample {sample.internal_id} to {validation_case_data.validation_case.internal_id}"
76	)
77
78	@staticmethod	1✔
79	def _get_sample_status(sample: Sample) -> str:	1✔
80	"""Return the status of a sample."""
81	return sample.links[0].status if sample.links else "unknown"	1✔
82
83	def create_validation_case_in_statusdb(self, validation_case_data: ValidationCaseData) -> None:	1✔
84	"""
85	Add the validation samples and case to statusDB and generate the sample case link.
86	"""
87	try:	1✔
88	self.store_validation_samples(validation_case_data)	1✔
89	self.store_validation_case(validation_case_data)	1✔
90	self._link_sample_to_case(validation_case_data)	1✔
91	self.status_db.session.commit()	1✔
NEW 92	except Exception as error:	×
NEW 93	LOG.warning(f"An error occurred {repr(error)}")	×
NEW 94	self.status_db.session.rollback()	×
95
96	def create_validation_sample_bundles(self, validation_samples: list[Sample]) -> None:	1✔
97	for validation_sample in validation_samples:	1✔
98	self.hk_api.create_new_bundle_and_version(validation_sample.internal_id)	1✔
99
100	def copy_original_sample_bundle_files(self, validation_samples: list[Sample]) -> None:	1✔
101	"""Copy the fastq files from the original sample bundle into the new validation sample bundle"""
102	for validation_sample in validation_samples:	1✔
103	fastq_files: list[File] = self.hk_api.get_files_from_latest_version(	1✔
104	bundle_name=validation_sample.from_sample, tags=[SequencingFileTag.FASTQ]
105	)
106	validation_bundle_path: Path = self.hk_api.get_latest_bundle_version(	1✔
107	bundle_name=validation_sample.internal_id
108	).full_path
109	if not validation_bundle_path.exists():	1✔
110	os.makedirs(validation_bundle_path)	1✔
111	for fastq_file in fastq_files:	1✔
112	copy_file(file_path=Path(fastq_file.full_path), destination=validation_bundle_path)	1✔
113
114	def rename_and_add_original_sample_bundle_files(self, validation_samples: list[Sample]) -> None:	1✔
115	"""Rename the fastq files to use the new validation sample internal id."""
116	for validation_sample in validation_samples:	1✔
117	validation_bundle_path: Path = self.hk_api.get_latest_bundle_version(	1✔
118	bundle_name=validation_sample.internal_id
119	).full_path
120	fastq_files: list[Path] = get_files_matching_pattern(	1✔
121	directory=validation_bundle_path, pattern="." + SequencingFileTag.FASTQ + "."
122	)
123	for fastq_file in fastq_files:	1✔
124	new_fast_file_path = self.get_new_fastq_file_path(	1✔
125	fastq_file=fastq_file, validation_sample=validation_sample
126	)
127	rename_file(file_path=fastq_file, renamed_file_path=new_fast_file_path)	1✔
128	new_tags: list[str] = self.get_new_tags(validation_sample)	1✔
129	self.hk_api.add_and_include_file_to_latest_version(	1✔
130	bundle_name=validation_sample.internal_id,
131	file=new_fast_file_path,
132	tags=new_tags,
133	)
134
135	def get_new_tags(self, validation_sample: Sample) -> list[str]:	1✔
136	"""
137	Get the fastq file tags from the original sample and
138	replace the bundle name with the validation sample internal id.
139	"""
140	hk_files: list[File] = self.hk_api.get_files_from_latest_version(	1✔
141	bundle_name=validation_sample.from_sample, tags=[SequencingFileTag.FASTQ]
142	)
143	original_tags: list[str] = self.hk_api.get_tag_names_from_file(hk_files[0])	1✔
144	new_tags: list[str] = [validation_sample.internal_id, VALIDATION_TAG]	1✔
145	for tag in original_tags:	1✔
146	if tag != validation_sample.from_sample:	1✔
147	new_tags.append(tag)	1✔
148	return new_tags	1✔
149
150	@staticmethod	1✔
151	def get_new_fastq_file_path(fastq_file: Path, validation_sample: Sample) -> Path:	1✔
152	return Path(	1✔
153	fastq_file.as_posix().replace(
154	validation_sample.from_sample,
155	validation_sample.internal_id,
156	)
157	)
158
159	def create_validation_samples_in_housekeeper(self, validation_case_data: ValidationCaseData):	1✔
160	self.create_validation_sample_bundles(validation_case_data.validation_samples)	1✔
161	self.copy_original_sample_bundle_files(validation_case_data.validation_samples)	1✔
162	self.rename_and_add_original_sample_bundle_files(validation_case_data.validation_samples)	1✔
163
164	def create_validation_case(	1✔
165	self,
166	case_id: str,
167	case_name: str,
168	data_analysis: str \| None = None,
169	delivery: str \| None = None,
170	) -> None:
171	"""Create the validation case in statusdb and associated sample bundles in housekeeper."""
172	validation_data_input = ValidationDataInput(	1✔
173	case_id=case_id, case_name=case_name, data_analysis=data_analysis, delivery=delivery
174	)
175	validation_case_data: ValidationCaseData = self.get_validation_case_data(	1✔
176	validation_data_input
177	)
178	self.create_validation_case_in_statusdb(validation_case_data)	1✔
179	self.create_validation_samples_in_housekeeper(validation_case_data)	1✔

Clinical-Genomics / cg / 8892829037

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