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

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 _slicingDuration: Nullable[float] #: Duration for sorting files by size and assigning them to Docker image layers. 

124 

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

126 self._root = root 

127 self._layers = [] 

128 

129 @readonly 

130 def Root(self) -> Root: 

131 return self._root 

132 

133 @readonly 

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

135 return self._layers 

136 

137 @readonly 

138 def LayerCount(self) -> int: 

139 return len(self._layers) 

140 

141 @readonly 

142 def TotalFileCount(self) -> int: 

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

144 

145 @readonly 

146 def SlicingDuration(self) -> float: 

147 """ 

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

149 

150 :returns: The slicing duration in seconds. 

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

152 """ 

153 if self._slicingDuration is None: 

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

155 

156 return self._slicingDuration 

157 

158 def CreateDockerLayers( 

159 self, 

160 minLayerSize: int, 

161 maxLayerSize: int, 

162 layerSizeGradient: int 

163 ) -> List[Layer]: 

164 layer = Layer(self) 

165 

166 collectedFiles = set() 

167 targetLayerSize = maxLayerSize 

168 

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

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

171 

172 with Stopwatch() as sw: 

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

174 firstFile = next(iterator) 

175 collectedFiles |= layer.AddFile(firstFile) 

176 

177 for file in iterator: 

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

179 continue 

180 

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

182 collectedFiles |= layer.AddFile(file) 

183 else: 

184 layer = Layer(self, layer) 

185 collectedFiles |= layer.AddFile(file) 

186 

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

188 targetLayerSize = size 

189 

190 self._slicingDuration = sw.Duration 

191 

192 def WriteLayerFiles(self, directory: Path, relative: bool = True) -> None: 

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

194 layer.WriteLayerFile(directory / f"layer_{i}.files", relative)