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

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

65 Read-only property to access the layer cake this layer belongs to (:attr:`_parent`). 

66 

67 :returns: The layer cake this layer belongs to, or ``None`` if the layer isn't part of one. 

68 """ 

69 return self._parent 

70 

71 @readonly 

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

73 """ 

74 Read-only property to access the layer below this one (:attr:`_previousLayer`). 

75 

76 :returns: The previous layer, or ``None`` if this is the bottom layer. 

77 """ 

78 return self._previousLayer 

79 

80 @readonly 

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

82 """ 

83 Read-only property to access the layer above this one (:attr:`_nextLayer`). 

84 

85 :returns: The next layer, or ``None`` if this is the top layer. 

86 """ 

87 return self._nextLayer 

88 

89 @readonly 

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

91 """ 

92 Read-only property to access the files contributed by this layer (:attr:`_files`). 

93 

94 :returns: List of files in this layer. 

95 """ 

96 return self._files 

97 

98 @readonly 

99 def FileCount(self) -> int: 

100 """ 

101 Read-only property to return the number of files in this layer. 

102 

103 :returns: Number of files. 

104 """ 

105 return len(self._files) 

106 

107 @readonly 

108 def Size(self) -> int: 

109 """ 

110 Read-only property to access the accumulated size of all files in this layer (:attr:`_size`). 

111 

112 :returns: Size of the layer in bytes. 

113 """ 

114 return self._size 

115 

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

117 usedFiles = set() 

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

119 for filename in element.File.Parents: 

120 self._files.append(filename) 

121 usedFiles.add(filename) 

122 elif isinstance(element, SymbolicLink): 

123 self._files.append(element) 

124 usedFiles.add(element) 

125 else: 

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

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

128 raise ex 

129 

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

131 

132 return usedFiles 

133 

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

135 rootDirectory = self._parent._root._path 

136 

137 if relative: 

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

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

140 else: 

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

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

143 

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

145 for file in self._files: 

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

147 

148 

149@export 

150class LayerCake(metaclass=ExtendedType): 

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

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

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

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

155 

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

157 self._root = root 

158 self._layers = [] 

159 self._emptyDirectories = [] 

160 

161 @readonly 

162 def Root(self) -> Root: 

163 """ 

164 Read-only property to access the root directory of the merged layers (:attr:`_root`). 

165 

166 :returns: Root directory of the layer cake. 

167 """ 

168 return self._root 

169 

170 @readonly 

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

172 """ 

173 Read-only property to access all layers, bottom-most first (:attr:`_layers`). 

174 

175 :returns: List of layers. 

176 """ 

177 return self._layers 

178 

179 @readonly 

180 def LayerCount(self) -> int: 

181 """ 

182 Read-only property to return the number of layers. 

183 

184 :returns: Number of layers. 

185 """ 

186 return len(self._layers) 

187 

188 @readonly 

189 def TotalFileCount(self) -> int: 

190 """ 

191 Read-only property to return the number of files across all layers. 

192 

193 :returns: Sum of all layers' file counts. 

194 """ 

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

196 

197 @readonly 

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

199 """ 

200 Read-only property to access the directories that contain no files (:attr:`_emptyDirectories`). 

201 

202 :returns: List of empty directories. 

203 """ 

204 return self._emptyDirectories 

205 

206 @readonly 

207 def EmptyDirectoryCount(self) -> int: 

208 """ 

209 Read-only property to return the number of empty directories. 

210 

211 :returns: Number of empty directories. 

212 """ 

213 return len(self._emptyDirectories) 

214 

215 @readonly 

216 def SlicingDuration(self) -> float: 

217 """ 

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

219 

220 :returns: The slicing duration in seconds. 

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

222 """ 

223 if self._slicingDuration is None: 

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

225 

226 return self._slicingDuration 

227 

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

229 with Stopwatch() as sw: 

230 self._SliceFilesystemIntoLayers(minLayerSize, maxLayerSize, layerSizeGradient) 

231 self._CollectEmptDirectories() 

232 

233 self._slicingDuration = sw.Duration 

234 

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

236 # greedy algorithm 

237 layer = Layer(self) 

238 

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

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

241 

242 collectedFiles = set() 

243 targetLayerSize = maxLayerSize 

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

245 firstFile = next(iterator) 

246 collectedFiles |= layer.AddFile(firstFile) 

247 

248 for file in iterator: 

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

250 continue 

251 

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

253 collectedFiles |= layer.AddFile(file) 

254 else: 

255 layer = Layer(self, layer) 

256 collectedFiles |= layer.AddFile(file) 

257 

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

259 targetLayerSize = size 

260 

261 def _CollectEmptDirectories(self) -> None: 

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

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

264 self._emptyDirectories.append(directory) 

265 

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

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

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

269 

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

271 rootDirectory = self._root._path 

272 

273 if relative: 

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

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

276 else: 

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

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

279 

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

281 for directory in self._emptyDirectories: 

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