Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/exporter/_abc_exporter.py: 75%
46 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:32 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:32 +0000
1import zipfile 1a
2from abc import abstractmethod, abstractproperty 1a
3from collections.abc import Callable, Iterator 1a
4from dataclasses import dataclass 1a
5from pathlib import Path 1a
6from uuid import UUID 1a
8from pydantic import BaseModel 1a
10from mealie.core.root_logger import get_logger 1a
11from mealie.repos.all_repositories import AllRepositories 1a
12from mealie.schema.reports.reports import ReportEntryCreate 1a
14from .._base_service import BaseService 1a
17@dataclass 1a
18class ExportedItem: 1a
19 """
20 Exported items are the items provided by items() call in an concrete exporter class
21 Where the items are used to write data to the zip file. Models should derive from the
22 BaseModel class OR provide a .json method that returns a json string.
23 """
25 model: BaseModel 1a
26 name: str 1a
29class ABCExporter(BaseService): 1a
30 write_dir_to_zip: Callable[[Path, str, set[str] | None], None] | None = None 1a
32 def __init__(self, db: AllRepositories, group_id: UUID) -> None: 1a
33 self.logger = get_logger() 1bc
34 self.db = db 1bc
35 self.group_id = group_id 1bc
37 super().__init__() 1bc
39 @abstractproperty 1a
40 def destination_dir(self) -> str: ... 40 ↛ exitline 40 didn't return from function 'destination_dir' because 1a
42 @abstractmethod 1a
43 def items(self) -> Iterator[ExportedItem]: ... 43 ↛ exitline 43 didn't return from function 'items' because 1a
45 def _post_export_hook(self, _: BaseModel) -> None: 1a
46 pass
48 def export(self, zip: zipfile.ZipFile) -> list[ReportEntryCreate]: # type: ignore 1a
49 """
50 Export takes in a zip file and exports the recipes to it. Note that the zip
51 file open/close is NOT handled by this method. You must handle it yourself.
53 Args:
54 zip (zipfile.ZipFile): Zip file destination
56 Returns:
57 list[ReportEntryCreate]:
58 """
59 self.write_dir_to_zip = self.write_dir_to_zip_func(zip) 1bc
61 for item in self.items(): 1bc
62 if item is None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true1b
63 self.logger.error("Failed to export item. no item found")
64 continue
66 zip.writestr(f"{self.destination_dir}/{item.name}/{item.name}.json", item.model.model_dump_json()) 1b
68 self._post_export_hook(item.model) 1b
70 self.write_dir_to_zip = None 1bc
72 def write_dir_to_zip_func(self, zip: zipfile.ZipFile): 1a
73 """Returns a recursive function that writes a directory to a zip file.
75 Args:
76 zip (zipfile.ZipFile):
77 """
79 def func(source_dir: Path, dest_dir: str, ignore_ext: set[str] | None = None) -> None: 1bc
80 ignore_ext = ignore_ext or set() 1b
82 for source_file in source_dir.iterdir(): 82 ↛ 83line 82 didn't jump to line 83 because the loop on line 82 never started1b
83 if source_file.is_dir():
84 func(source_file, f"{dest_dir}/{source_file.name}")
85 elif source_file.suffix not in ignore_ext:
86 zip.write(source_file, f"{dest_dir}/{source_file.name}")
88 return func 1bc