Coverage for documents/bulk_download.py: 46%

1from __future__ import annotations 

2 

3from pathlib import Path 

4from typing import TYPE_CHECKING 

5from typing import NoReturn 

6 

7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8 because the condition on line 7 was never true

8 from collections.abc import Callable 

9 from zipfile import ZipFile 

10 

11 from documents.models import Document 

12 

13 

14class BulkArchiveStrategy: 

15 def __init__(self, zipf: ZipFile, *, follow_formatting: bool = False) -> None: 

16 self.zipf: ZipFile = zipf 

17 if follow_formatting: 

18 self.make_unique_filename: Callable[..., Path | str] = ( 

19 self._formatted_filepath 

20 ) 

21 else: 

22 self.make_unique_filename = self._filename_only 

23 

24 def _filename_only( 

25 self, 

26 doc: Document, 

27 *, 

28 archive: bool = False, 

29 folder: str = "", 

30 ) -> str: 

31 """ 

32 Constructs a unique name for the given document to be used inside the 

33 zip file. 

34 

35 The filename might not be unique enough, so a counter is appended if needed 

36 """ 

37 counter = 0 

38 while True: 

39 filename: str = folder + doc.get_public_filename( 

40 archive=archive, 

41 counter=counter, 

42 ) 

43 if filename in self.zipf.namelist(): 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true

44 counter += 1 

45 else: 

46 return filename 

47 

48 def _formatted_filepath( 

49 self, 

50 doc: Document, 

51 *, 

52 archive: bool = False, 

53 folder: str = "", 

54 ) -> Path: 

55 """ 

56 Constructs a full file path for the given document to be used inside 

57 the zipfile. 

58 

59 The path is already unique, as handled when a document is consumed or updated 

60 """ 

61 if archive and doc.has_archive_version: 

62 if TYPE_CHECKING: 

63 assert doc.archive_filename is not None 

64 in_archive_path: Path = Path(folder) / doc.archive_filename 

65 else: 

66 if TYPE_CHECKING: 

67 assert doc.filename is not None 

68 in_archive_path = Path(folder) / doc.filename 

69 

70 return in_archive_path 

71 

72 def add_document(self, doc: Document) -> NoReturn: 

73 raise NotImplementedError # pragma: no cover 

74 

75 

76class OriginalsOnlyStrategy(BulkArchiveStrategy): 

77 def add_document(self, doc: Document) -> None: 

78 self.zipf.write(doc.source_path, self.make_unique_filename(doc)) 

79 

80 

81class ArchiveOnlyStrategy(BulkArchiveStrategy): 

82 def add_document(self, doc: Document) -> None: 

83 if doc.has_archive_version: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 if TYPE_CHECKING: 

85 assert doc.archive_path is not None 

86 self.zipf.write( 

87 doc.archive_path, 

88 self.make_unique_filename(doc, archive=True), 

89 ) 

90 else: 

91 self.zipf.write(doc.source_path, self.make_unique_filename(doc)) 

92 

93 

94class OriginalAndArchiveStrategy(BulkArchiveStrategy): 

95 def add_document(self, doc: Document) -> None: 

96 if doc.has_archive_version: 

97 if TYPE_CHECKING: 

98 assert doc.archive_path is not None 

99 self.zipf.write( 

100 doc.archive_path, 

101 self.make_unique_filename(doc, archive=True, folder="archive/"), 

102 ) 

103 

104 self.zipf.write( 

105 doc.source_path, 

106 self.make_unique_filename(doc, folder="originals/"), 

107 )