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 datetime import timedelta 

41from pathlib import Path 

42from typing import Optional as Nullable 

43 

44from pyTooling.Decorators import export, readonly 

45from pyTooling.MetaClasses import ExtendedType 

46from pyTooling.Common import getFullyQualifiedName 

47from pyTooling.Filesystem import Root, Element, Directory, Filename, SymbolicLink, FilesystemError 

48from pyTooling.Stopwatch import Stopwatch 

49 

50 

51@export 

52class Layer(metaclass=ExtendedType): 

53 """ 

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

55 

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

57 """ 

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

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

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

61 

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

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

64 

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

66 """ 

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

68 

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

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

71 """ 

72 if parent is not None: 

73 parent._layers.append(self) 

74 self._parent = parent 

75 self._previousLayer = previousLayer 

76 self._nextLayer = None 

77 if previousLayer is not None: 

78 previousLayer._nextLayer = self 

79 

80 self._files = [] 

81 self._size = 0 

82 

83 @readonly 

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

85 """ 

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

87 

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

89 """ 

90 return self._parent 

91 

92 @readonly 

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

94 """ 

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

96 

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

98 """ 

99 return self._previousLayer 

100 

101 @readonly 

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

103 """ 

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

105 

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

107 """ 

108 return self._nextLayer 

109 

110 @readonly 

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

112 """ 

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

114 

115 :returns: List of files in this layer. 

116 """ 

117 return self._files 

118 

119 @readonly 

120 def FileCount(self) -> int: 

121 """ 

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

123 

124 :returns: Number of files. 

125 """ 

126 return len(self._files) 

127 

128 @readonly 

129 def Size(self) -> int: 

130 """ 

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

132 

133 :returns: Size of the layer in bytes. 

134 """ 

135 return self._size 

136 

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

138 """ 

139 Add a filename or symbolic link to this layer. 

140 

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

142 the same image layer. 

143 

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

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

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

147 """ 

148 usedFiles = set() 

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

150 for filename in element.File.Parents: 

151 self._files.append(filename) 

152 usedFiles.add(filename) 

153 elif isinstance(element, SymbolicLink): 

154 self._files.append(element) 

155 usedFiles.add(element) 

156 else: 

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

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

159 raise ex 

160 

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

162 

163 return usedFiles 

164 

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

166 """ 

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

168 

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

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

171 """ 

172 rootDirectory = self._parent._root._path 

173 

174 if relative: 

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

176 """ 

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

178 

179 :param file: The path to render. 

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

181 """ 

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

183 else: 

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

185 """ 

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

187 

188 :param file: The path to render. 

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

190 """ 

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

192 

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

194 for file in self._files: 

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

196 

197 

198@export 

199class LayerCake(metaclass=ExtendedType): 

200 """ 

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

202 

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

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

205 directories no layer covers. 

206 """ 

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

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

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

210 _slicingDuration: Nullable[timedelta] #: Time to sort files by size and assign them to layers. 

211 

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

213 """ 

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

215 

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

217 """ 

218 self._root = root 

219 self._layers = [] 

220 self._emptyDirectories = [] 

221 

222 @readonly 

223 def Root(self) -> Root: 

224 """ 

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

226 

227 :returns: Root directory of the layer cake. 

228 """ 

229 return self._root 

230 

231 @readonly 

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

233 """ 

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

235 

236 :returns: List of layers. 

237 """ 

238 return self._layers 

239 

240 @readonly 

241 def LayerCount(self) -> int: 

242 """ 

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

244 

245 :returns: Number of layers. 

246 """ 

247 return len(self._layers) 

248 

249 @readonly 

250 def TotalFileCount(self) -> int: 

251 """ 

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

253 

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

255 """ 

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

257 

258 @readonly 

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

260 """ 

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

262 

263 :returns: List of empty directories. 

264 """ 

265 return self._emptyDirectories 

266 

267 @readonly 

268 def EmptyDirectoryCount(self) -> int: 

269 """ 

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

271 

272 :returns: Number of empty directories. 

273 """ 

274 return len(self._emptyDirectories) 

275 

276 @readonly 

277 def SlicingDuration(self) -> timedelta: 

278 """ 

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

280 

281 :returns: The slicing duration. 

282 :raises FilesystemError: If the filesystem was not sliced into layers. 

283 """ 

284 if self._slicingDuration is None: 

285 raise FilesystemError("Filesystem was not sliced, yet.") 

286 

287 return self._slicingDuration 

288 

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

290 """ 

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

292 

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

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

295 

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

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

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

299 """ 

300 with Stopwatch() as sw: 

301 self._SliceFilesystemIntoLayers(minLayerSize, maxLayerSize, layerSizeGradient) 

302 self._CollectEmptDirectories() 

303 

304 self._slicingDuration = sw.Duration 

305 

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

307 """ 

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

309 

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

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

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

313 """ 

314 # greedy algorithm 

315 layer = Layer(self) 

316 

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

318 """ 

319 Nested function used as sort key. 

320 

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

322 

323 :param file: The filesystem element to measure. 

324 :returns: Size of the element in bytes. 

325 """ 

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

327 

328 collectedFiles = set() 

329 targetLayerSize = maxLayerSize 

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

331 firstFile = next(iterator) 

332 collectedFiles |= layer.AddFile(firstFile) 

333 

334 for file in iterator: 

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

336 continue 

337 

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

339 collectedFiles |= layer.AddFile(file) 

340 else: 

341 layer = Layer(self, layer) 

342 collectedFiles |= layer.AddFile(file) 

343 

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

345 targetLayerSize = size 

346 

347 def _CollectEmptDirectories(self) -> None: 

348 """ 

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

350 """ 

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

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

353 self._emptyDirectories.append(directory) 

354 

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

356 """ 

357 Write one file list per layer. 

358 

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

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

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

362 """ 

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

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

365 

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

367 """ 

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

369 

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

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

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

373 """ 

374 rootDirectory = self._root._path 

375 

376 if relative: 

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

378 """ 

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

380 

381 :param file: The path to render. 

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

383 """ 

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

385 else: 

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

387 """ 

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

389 

390 :param file: The path to render. 

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

392 """ 

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

394 

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

396 for directory in self._emptyDirectories: 

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