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

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

7 

8from pydantic import BaseModel 1a

9 

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

13 

14from .._base_service import BaseService 1a

15 

16 

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

24 

25 model: BaseModel 1a

26 name: str 1a

27 

28 

29class ABCExporter(BaseService): 1a

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

31 

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

36 

37 super().__init__() 1bc

38 

39 @abstractproperty 1a

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

41 

42 @abstractmethod 1a

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

44 

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

46 pass 

47 

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. 

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) 1bc

60 

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 

65 

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

67 

68 self._post_export_hook(item.model) 1b

69 

70 self.write_dir_to_zip = None 1bc

71 

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. 

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: 1bc

80 ignore_ext = ignore_ext or set() 1b

81 

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

87 

88 return func 1bc