Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/exporter/_abc_exporter.py: 62%

1import zipfile 

2from abc import abstractmethod, abstractproperty 

3from collections.abc import Callable, Iterator 

4from dataclasses import dataclass 

5from pathlib import Path 

6from uuid import UUID 

7 

8from pydantic import BaseModel 

9 

10from mealie.core.root_logger import get_logger 

11from mealie.repos.all_repositories import AllRepositories 

12from mealie.schema.reports.reports import ReportEntryCreate 

13 

14from .._base_service import BaseService 

15 

16 

17@dataclass 

18class ExportedItem: 

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 """ 

24 

25 model: BaseModel 

26 name: str 

27 

28 

29class ABCExporter(BaseService): 

30 write_dir_to_zip: Callable[[Path, str, set[str] | None], None] | None = None 

31 

32 def __init__(self, db: AllRepositories, group_id: UUID) -> None: 

33 self.logger = get_logger() 

34 self.db = db 

35 self.group_id = group_id 

36 

37 super().__init__() 

38 

39 @abstractproperty 

40 def destination_dir(self) -> str: ... 40 ↛ exitline 40 didn't return from function 'destination_dir' because

41 

42 @abstractmethod 

43 def items(self) -> Iterator[ExportedItem]: ... 43 ↛ exitline 43 didn't return from function 'items' because

44 

45 def _post_export_hook(self, _: BaseModel) -> None: 

46 pass 

47 

48 def export(self, zip: zipfile.ZipFile) -> list[ReportEntryCreate]: # type: ignore 

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. 

52 

53 Args: 

54 zip (zipfile.ZipFile): Zip file destination 

55 

56 Returns: 

57 list[ReportEntryCreate]: 

58 """ 

59 self.write_dir_to_zip = self.write_dir_to_zip_func(zip) 

60 

61 for item in self.items(): 61 ↛ 62line 61 didn't jump to line 62 because the loop on line 61 never started

62 if item is None: 

63 self.logger.error("Failed to export item. no item found") 

64 continue 

65 

66 zip.writestr(f"{self.destination_dir}/{item.name}/{item.name}.json", item.model.model_dump_json()) 

67 

68 self._post_export_hook(item.model) 

69 

70 self.write_dir_to_zip = None 

71 

72 def write_dir_to_zip_func(self, zip: zipfile.ZipFile): 

73 """Returns a recursive function that writes a directory to a zip file. 

74 

75 Args: 

76 zip (zipfile.ZipFile): 

77 """ 

78 

79 def func(source_dir: Path, dest_dir: str, ignore_ext: set[str] | None = None) -> None: 

80 ignore_ext = ignore_ext or set() 

81 

82 for source_file in source_dir.iterdir(): 

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}") 

87 

88 return func