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# 

31""" 

32Slice a filesystem tree into Docker image layers. 

33 

34A :class:`~pyTooling.Filesystem.Docker.LayerCake` distributes the files of a 

35:class:`~pyTooling.Filesystem.Root` over layers - largest file first, each layer filled up to a target size that 

36shrinks from layer to layer - and writes one file list per layer, ready to be turned into image layers. 

37""" 

38from __future__ import annotations 

39 

40from pathlib import Path 

41from typing import Optional as Nullable 

42 

43from pyTooling.Decorators import export, readonly 

44from pyTooling.MetaClasses import ExtendedType 

45from pyTooling.Common import getFullyQualifiedName 

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

47from pyTooling.Stopwatch import Stopwatch 

48 

49 

50@export 

51class Layer(metaclass=ExtendedType): 

52 """ 

53 One layer of a Docker image: the files assigned to it, and its neighboring layers in the layer cake. 

54 

55 A layer knows its aggregated size, so the slicing algorithm can stop filling it when the target size is reached. 

56 """ 

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

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

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

60 

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

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

63 

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

65 """ 

66 Initialize an empty layer, which appends itself to the layer cake it belongs to. 

67 

68 :param parent: Optional, layer cake this layer is part of. 

69 :param previousLayer: Optional, layer below this one, which is linked to this layer in both directions. 

70 """ 

71 if parent is not None: 

72 parent._layers.append(self) 

73 self._parent = parent 

74 self._previousLayer = previousLayer 

75 self._nextLayer = None 

76 if previousLayer is not None: 

77 previousLayer._nextLayer = self 

78 

79 self._files = [] 

80 self._size = 0 

81 

82 @readonly 

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

84 """ 

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

86 

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

88 """ 

89 return self._parent 

90 

91 @readonly 

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

93 """ 

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

95 

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

97 """ 

98 return self._previousLayer 

99 

100 @readonly 

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

102 """ 

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

104 

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

106 """ 

107 return self._nextLayer 

108 

109 @readonly 

110 def Files(self) -> list[Element[Directory]]: 

111 """ 

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

113 

114 :returns: List of files in this layer. 

115 """ 

116 return self._files 

117 

118 @readonly 

119 def FileCount(self) -> int: 

120 """ 

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

122 

123 :returns: Number of files. 

124 """ 

125 return len(self._files) 

126 

127 @readonly 

128 def Size(self) -> int: 

129 """ 

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

131 

132 :returns: Size of the layer in bytes. 

133 """ 

134 return self._size 

135 

136 def AddFile(self, element: Element) -> set[Filename]: 

137 """ 

138 Add a filename or symbolic link to this layer. 

139 

140 For a filename, every other filename of the same file object is added too, because hardlinks have to end up in 

141 the same image layer. 

142 

143 :param element: The filename or symbolic link to add. 

144 :returns: The set of elements that were added, so the caller can skip them. 

145 :raises TypeError: If parameter 'element' is neither a filename nor a symbolic link. 

146 """ 

147 usedFiles = set() 

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

149 for filename in element.File.Parents: 

150 self._files.append(filename) 

151 usedFiles.add(filename) 

152 elif isinstance(element, SymbolicLink): 

153 self._files.append(element) 

154 usedFiles.add(element) 

155 else: 

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

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

158 raise ex 

159 

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

161 

162 return usedFiles 

163 

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

165 """ 

166 Write the layer's files as one file list. 

167 

168 :param path: Path of the file list to write. 

169 :param relative: Optional, if ``True``, the paths are written relative to the filesystem root. 

170 """ 

171 rootDirectory = self._parent._root._path 

172 

173 if relative: 

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

175 """ 

176 Nested function rendering a file's path relative to the filesystem root. 

177 

178 :param file: The path to render. 

179 :returns: The relative path in POSIX notation, terminated by a newline. 

180 """ 

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

182 else: 

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

184 """ 

185 Nested function rendering a file's path as it is. 

186 

187 :param file: The path to render. 

188 :returns: The absolute path in POSIX notation, terminated by a newline. 

189 """ 

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

191 

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

193 for file in self._files: 

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

195 

196 

197@export 

198class LayerCake(metaclass=ExtendedType): 

199 """ 

200 A stack of Docker image layers computed from a filesystem tree. 

201 

202 :meth:`CreateDockerLayers` distributes the files of a :class:`~pyTooling.Filesystem.Root` over layers - largest file 

203 first, each layer filled up to a target size that shrinks by a gradient from layer to layer - and collects the 

204 directories no layer covers. 

205 """ 

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

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

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

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

210 

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

212 """ 

213 Initialize an empty layer cake for the given filesystem tree. 

214 

215 :param root: Root of the filesystem statistics scope to slice into layers. 

216 """ 

217 self._root = root 

218 self._layers = [] 

219 self._emptyDirectories = [] 

220 

221 @readonly 

222 def Root(self) -> Root: 

223 """ 

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

225 

226 :returns: Root directory of the layer cake. 

227 """ 

228 return self._root 

229 

230 @readonly 

231 def Layers(self) -> list[Layer]: 

232 """ 

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

234 

235 :returns: List of layers. 

236 """ 

237 return self._layers 

238 

239 @readonly 

240 def LayerCount(self) -> int: 

241 """ 

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

243 

244 :returns: Number of layers. 

245 """ 

246 return len(self._layers) 

247 

248 @readonly 

249 def TotalFileCount(self) -> int: 

250 """ 

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

252 

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

254 """ 

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

256 

257 @readonly 

258 def EmptyDirectories(self) -> list[Directory]: 

259 """ 

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

261 

262 :returns: List of empty directories. 

263 """ 

264 return self._emptyDirectories 

265 

266 @readonly 

267 def EmptyDirectoryCount(self) -> int: 

268 """ 

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

270 

271 :returns: Number of empty directories. 

272 """ 

273 return len(self._emptyDirectories) 

274 

275 @readonly 

276 def SlicingDuration(self) -> float: 

277 """ 

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

279 

280 :returns: The slicing duration in seconds. 

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

282 """ 

283 if self._slicingDuration is None: 

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

285 

286 return self._slicingDuration 

287 

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

289 """ 

290 Distribute the filesystem's files over image layers and collect the directories no layer covers. 

291 

292 The layers are filled largest file first. Each layer is filled up to a target size, which shrinks by 

293 ``layerSizeGradient`` from layer to layer until ``minLayerSize`` is reached. 

294 

295 :param minLayerSize: Smallest target size a layer is filled to. 

296 :param maxLayerSize: Target size of the first layer. 

297 :param layerSizeGradient: Amount the target size shrinks by from layer to layer. 

298 """ 

299 with Stopwatch() as sw: 

300 self._SliceFilesystemIntoLayers(minLayerSize, maxLayerSize, layerSizeGradient) 

301 self._CollectEmptDirectories() 

302 

303 self._slicingDuration = sw.Duration 

304 

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

306 """ 

307 Distribute the filesystem's files over image layers, largest file first. 

308 

309 :param minLayerSize: Smallest target size a layer is filled to. 

310 :param maxLayerSize: Target size of the first layer. 

311 :param layerSizeGradient: Amount the target size shrinks by from layer to layer. 

312 """ 

313 # greedy algorithm 

314 layer = Layer(self) 

315 

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

317 """ 

318 Nested function used as sort key. 

319 

320 A symbolic link occupies no space of its own, so it is counted as zero and ends up last. 

321 

322 :param file: The filesystem element to measure. 

323 :returns: Size of the element in bytes. 

324 """ 

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

326 

327 collectedFiles = set() 

328 targetLayerSize = maxLayerSize 

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

330 firstFile = next(iterator) 

331 collectedFiles |= layer.AddFile(firstFile) 

332 

333 for file in iterator: 

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

335 continue 

336 

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

338 collectedFiles |= layer.AddFile(file) 

339 else: 

340 layer = Layer(self, layer) 

341 collectedFiles |= layer.AddFile(file) 

342 

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

344 targetLayerSize = size 

345 

346 def _CollectEmptDirectories(self) -> None: 

347 """ 

348 Collect the directories that contain neither files nor subdirectories, so no layer covers them. 

349 """ 

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

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

352 self._emptyDirectories.append(directory) 

353 

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

355 """ 

356 Write one file list per layer. 

357 

358 :param directory: Directory the file lists are written to. 

359 :param fileNamePattern: Optional, pattern of the file names, with ``{layerID}`` replaced by the layer's number. 

360 :param relative: Optional, if ``True``, the paths are written relative to the filesystem root. 

361 """ 

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

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

364 

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

366 """ 

367 Write the empty directories as one file list, so an image build can recreate them. 

368 

369 :param directory: Directory the file list is written to. 

370 :param fileNamePattern: Optional, name of the file list to write. 

371 :param relative: Optional, if ``True``, the paths are written relative to the filesystem root. 

372 """ 

373 rootDirectory = self._root._path 

374 

375 if relative: 

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

377 """ 

378 Nested function rendering a directory's path relative to the filesystem root. 

379 

380 :param file: The path to render. 

381 :returns: The relative path in POSIX notation, terminated by a newline. 

382 """ 

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

384 else: 

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

386 """ 

387 Nested function rendering a directory's path as it is. 

388 

389 :param file: The path to render. 

390 :returns: The absolute path in POSIX notation, terminated by a newline. 

391 """ 

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

393 

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

395 for directory in self._emptyDirectories: 

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