10504067794

Committed 22 Aug 2024 07:54AM UTC coverage: 84.772%. First build

Build # 10504067794

Build Type

Pull #3622

github

Committed by

web-flow

Commit Message

Merge 9063fc86c into 44bbe281f

Pull Request Pull Request #3622: feat(delivery file mover service)

Run Details

49 of 51 new or added lines in 5 files covered. (96.08%)

21989 of 25939 relevant lines covered (84.77%)

0.85 hits per line

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

97.5

/cg/services/file_delivery/move_files_service/move_delivery_files_service.py

from pathlib import Path
from cg.constants.delivery import INBOX_NAME
from cg.services.file_delivery.fetch_file_service.models import (
    DeliveryFiles,
    DeliveryMetaData,
    CaseFile,
    SampleFile,
)
from cg.utils.files import link_or_overwrite_file


class MoveDeliveryFilesService:
    """
    Class that encapsulates the logic required for moving files to the customer folder.
    """

    def move_files(self, delivery_files: DeliveryFiles, delivery_base_path: Path) -> DeliveryFiles:
        """Move the files to the customer folder."""
        inbox_path: Path = self._create_ticket_inbox_path(
            delivery_base_path, delivery_files.delivery_data
        )
        self._create_ticket_inbox_folder(inbox_path)
        self._create_hard_links_for_delivery_files(
            delivery_files=delivery_files, inbox_path=inbox_path
        )
        return self._replace_file_paths_with_inbox_paths(
            delivery_files=delivery_files, inbox_path=inbox_path
        )

    @staticmethod
    def _create_ticket_inbox_folder(
        inbox_path: Path,
    ) -> Path:
        """Create a ticket inbox folder in the customer folder."""
        if not inbox_path.exists():
            inbox_path.mkdir(parents=True)
            return inbox_path
        raise FileExistsError(f"Ticket inbox folder {inbox_path} already exists for ticket.")

    @staticmethod
    def _create_ticket_inbox_path(
        delivery_base_path: Path, delivery_data: DeliveryMetaData
    ) -> Path:
        """Create the path to the ticket inbox folder."""
        return Path(
            delivery_base_path,
            delivery_data.customer_internal_id,
            INBOX_NAME,
            delivery_data.ticket_id,
        )

    @staticmethod
    def _create_inbox_file_path(file_path: Path, inbox_path: Path) -> Path:
        """Create the path to the inbox file."""
        return Path(inbox_path, file_path.name)

    def _create_hard_link_file_paths(
        self, file_models: list[SampleFile | CaseFile], inbox_path: Path
    ) -> None:
        """Create hard links to the sample files in the customer folder."""
        for file_model in file_models:
            inbox_file_path: Path = self._create_inbox_file_path(
                file_path=file_model.file_path, inbox_path=inbox_path
            )
            link_or_overwrite_file(src=file_model.file_path, dst=inbox_file_path)

    def _create_hard_links_for_delivery_files(
        self, delivery_files: DeliveryFiles, inbox_path: Path
    ) -> None:
        """Create hard links to the files in the customer folder."""
        if delivery_files.case_files:
            self._create_hard_link_file_paths(
                file_models=delivery_files.case_files, inbox_path=inbox_path
            )
        self._create_hard_link_file_paths(
            file_models=delivery_files.sample_files, inbox_path=inbox_path
        )

    def _replace_file_path_with_inbox_path(
        self, file_models: list[SampleFile | CaseFile], inbox_path: Path
    ) -> list[SampleFile | CaseFile]:
        """Replace the file path with the inbox path."""
        for file_model in file_models:
            inbox_file_path: Path = self._create_inbox_file_path(
                file_path=file_model.file_path, inbox_path=inbox_path
            )
            file_model.file_path = inbox_file_path
        return file_models

    def _replace_file_paths_with_inbox_paths(
        self,
        delivery_files: DeliveryFiles,
        inbox_path: Path,
    ) -> DeliveryFiles:
        """
        Replace to original file paths in the delivery files with the customer inbox file paths.
        """
        if delivery_files.case_files:
            delivery_files.case_files = self._replace_file_path_with_inbox_path(
                file_models=delivery_files.case_files, inbox_path=inbox_path
            )
        delivery_files.sample_files = self._replace_file_path_with_inbox_path(
            file_models=delivery_files.sample_files, inbox_path=inbox_path
        )
        return delivery_files

1	from pathlib import Path	1✔
2	from cg.constants.delivery import INBOX_NAME	1✔
3	from cg.services.file_delivery.fetch_file_service.models import (	1✔
4	DeliveryFiles,
5	DeliveryMetaData,
6	CaseFile,
7	SampleFile,
8	)
9	from cg.utils.files import link_or_overwrite_file	1✔
10
11
12	class MoveDeliveryFilesService:	1✔
13	"""
14	Class that encapsulates the logic required for moving files to the customer folder.
15	"""
16
17	def move_files(self, delivery_files: DeliveryFiles, delivery_base_path: Path) -> DeliveryFiles:	1✔
18	"""Move the files to the customer folder."""
19	inbox_path: Path = self._create_ticket_inbox_path(	1✔
20	delivery_base_path, delivery_files.delivery_data
21	)
22	self._create_ticket_inbox_folder(inbox_path)	1✔
23	self._create_hard_links_for_delivery_files(	1✔
24	delivery_files=delivery_files, inbox_path=inbox_path
25	)
26	return self._replace_file_paths_with_inbox_paths(	1✔
27	delivery_files=delivery_files, inbox_path=inbox_path
28	)
29
30	@staticmethod	1✔
31	def _create_ticket_inbox_folder(	1✔
32	inbox_path: Path,
33	) -> Path:
34	"""Create a ticket inbox folder in the customer folder."""
35	if not inbox_path.exists():	1✔
36	inbox_path.mkdir(parents=True)	1✔
37	return inbox_path	1✔
NEW 38	raise FileExistsError(f"Ticket inbox folder {inbox_path} already exists for ticket.")	×
39
40	@staticmethod	1✔
41	def _create_ticket_inbox_path(	1✔
42	delivery_base_path: Path, delivery_data: DeliveryMetaData
43	) -> Path:
44	"""Create the path to the ticket inbox folder."""
45	return Path(	1✔
46	delivery_base_path,
47	delivery_data.customer_internal_id,
48	INBOX_NAME,
49	delivery_data.ticket_id,
50	)
51
52	@staticmethod	1✔
53	def _create_inbox_file_path(file_path: Path, inbox_path: Path) -> Path:	1✔
54	"""Create the path to the inbox file."""
55	return Path(inbox_path, file_path.name)	1✔
56
57	def _create_hard_link_file_paths(	1✔
58	self, file_models: list[SampleFile \| CaseFile], inbox_path: Path
59	) -> None:
60	"""Create hard links to the sample files in the customer folder."""
61	for file_model in file_models:	1✔
62	inbox_file_path: Path = self._create_inbox_file_path(	1✔
63	file_path=file_model.file_path, inbox_path=inbox_path
64	)
65	link_or_overwrite_file(src=file_model.file_path, dst=inbox_file_path)	1✔
66
67	def _create_hard_links_for_delivery_files(	1✔
68	self, delivery_files: DeliveryFiles, inbox_path: Path
69	) -> None:
70	"""Create hard links to the files in the customer folder."""
71	if delivery_files.case_files:	1✔
72	self._create_hard_link_file_paths(	1✔
73	file_models=delivery_files.case_files, inbox_path=inbox_path
74	)
75	self._create_hard_link_file_paths(	1✔
76	file_models=delivery_files.sample_files, inbox_path=inbox_path
77	)
78
79	def _replace_file_path_with_inbox_path(	1✔
80	self, file_models: list[SampleFile \| CaseFile], inbox_path: Path
81	) -> list[SampleFile \| CaseFile]:
82	"""Replace the file path with the inbox path."""
83	for file_model in file_models:	1✔
84	inbox_file_path: Path = self._create_inbox_file_path(	1✔
85	file_path=file_model.file_path, inbox_path=inbox_path
86	)
87	file_model.file_path = inbox_file_path	1✔
88	return file_models	1✔
89
90	def _replace_file_paths_with_inbox_paths(	1✔
91	self,
92	delivery_files: DeliveryFiles,
93	inbox_path: Path,
94	) -> DeliveryFiles:
95	"""
96	Replace to original file paths in the delivery files with the customer inbox file paths.
97	"""
98	if delivery_files.case_files:	1✔
99	delivery_files.case_files = self._replace_file_path_with_inbox_path(	1✔
100	file_models=delivery_files.case_files, inbox_path=inbox_path
101	)
102	delivery_files.sample_files = self._replace_file_path_with_inbox_path(	1✔
103	file_models=delivery_files.sample_files, inbox_path=inbox_path
104	)
105	return delivery_files	1✔

Clinical-Genomics / cg / 10504067794

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