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

46 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:48 +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() 

34 self.db = db 

35 self.group_id = group_id 

36 

37 super().__init__() 

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) 

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): 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: 

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