Coverage for pyTooling / Filesystem / Docker.py: 71%

1# ==================================================================================================================== # 

2# _____ _ _ _____ _ _ _ # 

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ | ___(_) | ___ ___ _ _ ___| |_ ___ _ __ ___ # 

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` | | |_ | | |/ _ \/ __| | | / __| __/ _ \ '_ ` _ \ # 

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_| _| | | | __/\__ \ |_| \__ \ || __/ | | | | | # 

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_| |_|_|\___||___/\__, |___/\__\___|_| |_| |_| # 

7# |_| |___/ |___/ |___/ # 

8# ==================================================================================================================== # 

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

13# ==================================================================================================================== # 

14# Copyright 2025-2026 Patrick Lehmann - Bötzingen, Germany # 

15# # 

16# Licensed under the Apache License, Version 2.0 (the "License"); # 

17# you may not use this file except in compliance with the License. # 

18# You may obtain a copy of the License at # 

19# # 

20# http://www.apache.org/licenses/LICENSE-2.0 # 

21# # 

22# Unless required by applicable law or agreed to in writing, software # 

23# distributed under the License is distributed on an "AS IS" BASIS, # 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

25# See the License for the specific language governing permissions and # 

26# limitations under the License. # 

27# # 

28# SPDX-License-Identifier: Apache-2.0 # 

29# ==================================================================================================================== # 

30# 

31from pathlib import Path 

32from typing import Optional as Nullable, List, Set 

33 

34from pyTooling.Decorators import export, readonly 

35from pyTooling.MetaClasses import ExtendedType 

36from pyTooling.Common import getFullyQualifiedName 

37from pyTooling.Filesystem import Root, Element, Directory, Filename, SymbolicLink, FilesystemException 

38from pyTooling.Stopwatch import Stopwatch 

39 

40 

41@export 

42class Layer(metaclass=ExtendedType): 

43 _parent: Nullable["LayerCake"] #: Reference to the parent layer cake. 

44 _previousLayer: Nullable["Layer"] #: Reference to the previous layer. 

45 _nextLayer: Nullable["Layer"] #: Reference to the next layer 

46 

47 _files: List[Element[Directory]] #: List of files in this layer. 

48 _size: int #: Aggregated size of all contained files for this layer. 

49 

50 def __init__(self, parent: Nullable["LayerCake"] = None, previousLayer: Nullable["Layer"] = None) -> None: 

51 if parent is not None: 

52 parent._layers.append(self) 

53 self._parent = parent 

54 self._previousLayer = previousLayer 

55 self._nextLayer = None 

56 if previousLayer is not None: 

57 previousLayer._nextLayer = self 

58 

59 self._files = [] 

60 self._size = 0 

61 

62 @readonly 

63 def Parent(self) -> Nullable["LayerCake"]: 

64 return self._parent 

65 

66 @readonly 

67 def PreviousLayer(self) -> Nullable["Layer"]: 

68 return self._previousLayer 

69 

70 @readonly 

71 def NextLayer(self) -> Nullable["Layer"]: 

72 return self._nextLayer 

73 

74 @readonly 

75 def Files(self) -> List[Element[Directory]]: 

76 return self._files 

77 

78 @readonly 

79 def FileCount(self) -> int: 

80 return len(self._files) 

81 

82 @readonly 

83 def Size(self) -> int: 

84 return self._size 

85 

86 def AddFile(self, element: Element) -> Set[Filename]: 

87 usedFiles = set() 

88 if isinstance(element, Filename): 88 ↛ 92line 88 didn't jump to line 92 because the condition on line 88 was always true

89 for filename in element.File.Parents: 

90 self._files.append(filename) 

91 usedFiles.add(filename) 

92 elif isinstance(element, SymbolicLink): 

93 self._files.append(element) 

94 usedFiles.add(element) 

95 else: 

96 ex = TypeError(f"Parameter 'element' is not a filename nor symbolic link.") 

97 ex.add_note(f"Got type '{getFullyQualifiedName(element)}'.") 

98 raise ex 

99 

100 self._size += 0 if isinstance(element, SymbolicLink) else element.Size 

101 

102 return usedFiles 

103 

104 def WriteLayerFile(self, path: Path, relative: bool = True) -> None: 

105 rootDirectory = self._parent._root._path 

106 

107 if relative: 

108 def format(file: Path) -> str: 

109 return f"{file.relative_to(rootDirectory).as_posix()}\n" 

110 else: 

111 def format(file: Path) -> str: 

112 return f"{file.as_posix()}\n" 

113 

114 with path.open("w", encoding="utf-8") as f: 

115 for file in self._files: 

116 f.write(format(file.Path)) 

117 

118 

119@export 

120class LayerCake(metaclass=ExtendedType): 

121 _root: Nullable[Root] #: Reference to the filesystem root. 

122 _layers: List[Layer] #: List of Docker image layers. 

123 _emptyDirectories: List[Directory] #: List of empty directories (not covered by layers). 

124 _slicingDuration: Nullable[float] #: Duration for sorting files by size and assigning them to Docker image layers. 

125 

126 def __init__(self, root: Root) -> None: 

127 self._root = root 

128 self._layers = [] 

129 self._emptyDirectories = [] 

130 

131 @readonly 

132 def Root(self) -> Root: 

133 return self._root 

134 

135 @readonly 

136 def Layers(self) -> List[Layer]: 

137 return self._layers 

138 

139 @readonly 

140 def LayerCount(self) -> int: 

141 return len(self._layers) 

142 

143 @readonly 

144 def TotalFileCount(self) -> int: 

145 return sum(layer.FileCount for layer in self._layers) 

146 

147 @readonly 

148 def EmptyDirectories(self) -> List[Directory]: 

149 return self._emptyDirectories 

150 

151 @readonly 

152 def EmptyDirectoryCount(self) -> int: 

153 return len(self._emptyDirectories) 

154 

155 @readonly 

156 def SlicingDuration(self) -> float: 

157 """ 

158 Read-only property to access the time needed to slice the filesystem structure into docker layers. 

159 

160 :returns: The slicing duration in seconds. 

161 :raises FilesystemException: If the filesystem was not sliced into layers. 

162 """ 

163 if self._slicingDuration is None: 

164 raise FilesystemException(f"Filesystem was not sliced, yet.") 

165 

166 return self._slicingDuration 

167 

168 def CreateDockerLayers(self, minLayerSize: int, maxLayerSize: int, layerSizeGradient: int) -> None: 

169 with Stopwatch() as sw: 

170 self._SliceFilesystemIntoLayers(minLayerSize, maxLayerSize, layerSizeGradient) 

171 self._CollectEmptDirectories() 

172 

173 self._slicingDuration = sw.Duration 

174 

175 def _SliceFilesystemIntoLayers(self, minLayerSize: int, maxLayerSize: int, layerSizeGradient: int) -> None: 

176 # greedy algorithm 

177 layer = Layer(self) 

178 

179 def sizeOf(file: Element[Directory]) -> int: 

180 return 0 if isinstance(file, SymbolicLink) else file.Size 

181 

182 collectedFiles = set() 

183 targetLayerSize = maxLayerSize 

184 iterator = iter(sorted(self._root.IterateFiles(), key=sizeOf, reverse=True)) 

185 firstFile = next(iterator) 

186 collectedFiles |= layer.AddFile(firstFile) 

187 

188 for file in iterator: 

189 if file in collectedFiles: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true

190 continue 

191 

192 if layer._size + sizeOf(file) <= targetLayerSize: 

193 collectedFiles |= layer.AddFile(file) 

194 else: 

195 layer = Layer(self, layer) 

196 collectedFiles |= layer.AddFile(file) 

197 

198 if (size := targetLayerSize - layerSizeGradient) >= minLayerSize: 198 ↛ 188line 198 didn't jump to line 188 because the condition on line 198 was always true

199 targetLayerSize = size 

200 

201 def _CollectEmptDirectories(self) -> None: 

202 for directory in self._root.IterateDirectories(): 

203 if directory.SubdirectoryCount == 0 and directory.FileCount == 0: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true

204 self._emptyDirectories.append(directory) 

205 

206 def WriteLayerFiles(self, directory: Path, fileNamePattern: str = "layer_{layerID}.files", relative: bool = True) -> None: 

207 for i, layer in enumerate(self._layers, start=1): 

208 layer.WriteLayerFile(directory / fileNamePattern.format(layerID=i), relative) 

209 

210 def WriteEmptyDirectoryFile(self, directory: Path, fileNamePattern: str = "empty_directories.files", relative: bool = True) -> None: 

211 rootDirectory = self._root._path 

212 

213 if relative: 

214 def format(file: Path) -> str: 

215 return f"{file.relative_to(rootDirectory).as_posix()}\n" 

216 else: 

217 def format(file: Path) -> str: 

218 return f"{file.as_posix()}\n" 

219 

220 with (directory / fileNamePattern).open("w", encoding="utf-8") as f: 

221 for directory in self._emptyDirectories: 

222 f.write(format(directory.Path))