Coverage for pyTooling / Filesystem / __init__.py: 48%

462 statements  

« prev     ^ index     » next       coverage.py v7.12.0, created at 2025-11-21 22:22 +0000

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

2# _____ _ _ _____ _ _ _ # 

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

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

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

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

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

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

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

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

14# Copyright 2025-2025 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""" 

32An object-oriented file system abstraction for directory, file, symbolic link, ... statistics collection. 

33 

34.. important:: 

35 

36 This isn't a replacement of :mod:`pathlib` introduced with Python 3.4. 

37""" 

38from os import scandir, readlink 

39 

40from enum import Enum 

41from itertools import chain 

42from pathlib import Path 

43from typing import Optional as Nullable, Dict, Generic, Generator, TypeVar, List, Any, Callable, Union 

44 

45try: 

46 from pyTooling.Decorators import readonly, export 

47 from pyTooling.Exceptions import ToolingException 

48 from pyTooling.MetaClasses import ExtendedType, abstractmethod 

49 from pyTooling.Common import getFullyQualifiedName, zipdicts 

50 from pyTooling.Stopwatch import Stopwatch 

51 from pyTooling.Tree import Node 

52except (ImportError, ModuleNotFoundError): # pragma: no cover 

53 print("[pyTooling.Filesystem] Could not import from 'pyTooling.*'!") 

54 

55 try: 

56 from pyTooling.Decorators import readonly, export 

57 from pyTooling.Exceptions import ToolingException 

58 from pyTooling.MetaClasses import ExtendedType, abstractmethod 

59 from pyTooling.Common import getFullyQualifiedName 

60 from pyTooling.Stopwatch import Stopwatch 

61 from pyTooling.Tree import Node 

62 except (ImportError, ModuleNotFoundError) as ex: # pragma: no cover 

63 print("[pyTooling.Filesystem] Could not import directly!") 

64 raise ex 

65 

66 

67_ParentType = TypeVar("_ParentType", bound="Element") 

68 

69 

70@export 

71class FilesystemException(ToolingException): 

72 """Base-exception of all exceptions raised by :mod:`pyTooling.Filesystem`.""" 

73 

74 

75@export 

76class NodeKind(Enum): 

77 """ 

78 Node kind for filesystem elements in a :ref:`tree <STRUCT/Tree>`. 

79 

80 This enumeration is used when converting the filesystem statistics tree to an instance of :mod:`pyTooling.Tree`. 

81 """ 

82 Directory = 0 #: Node represents a directory. 

83 File = 1 #: Node represents a regular file. 

84 SymbolicLink = 2 #: Node represents a symbolic link. 

85 

86 

87@export 

88class Base(metaclass=ExtendedType, slots=True): 

89 _root: Nullable["Root"] #: Reference to the root of the filesystem statistics scope. 

90 _size: Nullable[int] #: Actual or aggregated size of the filesystem element. 

91 

92 def __init__( 

93 self, 

94 root: Nullable["Root"], 

95 size: Nullable[int], 

96 ) -> None: 

97 self._root = root 

98 self._size = size 

99 

100 @property 

101 def Root(self) -> Nullable["Root"]: 

102 """ 

103 Property to access the root of the filesystem statistics scope. 

104 

105 :returns: Root of the filesystem statistics scope. 

106 """ 

107 return self._root 

108 

109 @Root.setter 

110 def Root(self, value: "Root") -> None: 

111 self._root = value 

112 

113 @readonly 

114 def Size(self) -> int: 

115 """ 

116 Read-only property to access the elements size in Bytes. 

117 

118 :returns: Size in Bytes. 

119 :raises FilesystemException: If size is not computed, yet. 

120 """ 

121 if self._size is None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 raise FilesystemException("Size is not computed, yet.") 

123 

124 return self._size 

125 

126 # @abstractmethod 

127 def ToTree(self) -> Node: 

128 pass 

129 

130 

131@export 

132class Element(Base, Generic[_ParentType]): 

133 _name: str #: Name of the filesystem element. 

134 _parent: _ParentType #: Reference to the filesystem element's parent (:class:`Directory`) 

135 _linkSources: List["SymbolicLink"] #: A list of symbolic links pointing to this filesystem element. 

136 

137 def __init__( 

138 self, 

139 name: str, 

140 size: Nullable[int] = None, 

141 parent: Nullable[_ParentType] = None 

142 ) -> None: 

143 root = None # if parent is None else parent._root 

144 

145 super().__init__(root, size) 

146 

147 self._parent = parent 

148 self._name = name 

149 self._linkSources = [] 

150 

151 @property 

152 def Parent(self) -> _ParentType: 

153 return self._parent 

154 

155 @Parent.setter 

156 def Parent(self, value: _ParentType) -> None: 

157 self._parent = value 

158 

159 if value._root is not None: 

160 self._root = value._root 

161 

162 @readonly 

163 def Name(self) -> str: 

164 """ 

165 Read-only property to access the elements name. 

166 

167 :returns: Element name. 

168 """ 

169 return self._name 

170 

171 @readonly 

172 def Path(self) -> Path: 

173 raise NotImplemented(f"Property 'Path' is abstract.") 

174 

175 def AddLinkSources(self, source: "SymbolicLink") -> None: 

176 self._linkSources.append(source) 

177 

178 

179@export 

180class Directory(Element["Directory"]): 

181 """ 

182 A **directory** represents a directory in the filesystem contains subdirectories, regular files and symbolic links. 

183 

184 While scanning for subelements, the directory is populated with elements. Every file object added, gets registered in 

185 the filesystems :class:`Root` for deduplication. In case a file identifier already exists, the found filename will 

186 reference the same file objects. In turn, the file objects has then references to multiple filenames (parents). This 

187 allows to detect :term:`hardlinks <hardlink>`. 

188 

189 The time needed for scanning the directory and its subelements is provided via :data:`ScanDuration`. 

190 

191 After scnaning the directory for subelements, certain directory properties get aggregated. The time needed for 

192 aggregation is provided via :data:`AggregateDuration`. 

193 """ 

194 

195 _path: Nullable[Path] #: Cached :class:`~pathlib.Path` object of this directory. 

196 _subdirectories: Dict[str, "Directory"] #: Dictionary containing name-:class:`Directory` pairs. 

197 _files: Dict[str, "Filename"] #: Dictionary containing name-:class:`Filename` pairs. 

198 _symbolicLinks: Dict[str, "SymbolicLink"] #: Dictionary containing name-:class:`SymbolicLink` pairs. 

199 _collapsed: bool #: True, if this directory was collapsed. It contains no subelements. 

200 _scanDuration: Nullable[float] #: Duration for scanning the directory and all its subelements. 

201 _aggregateDuration: Nullable[float] #: Duration for aggregating all subelements. 

202 

203 def __init__( 

204 self, 

205 name: str, 

206 collectSubdirectories: bool = False, 

207 parent: Nullable["Directory"] = None 

208 ) -> None: 

209 super().__init__(name, None, parent) 

210 

211 self._path = None 

212 self._subdirectories = {} 

213 self._files = {} 

214 self._symbolicLinks = {} 

215 self._collapsed = False 

216 self._scanDuration = None 

217 self._aggregateDuration = None 

218 

219 if parent is not None: 

220 parent._subdirectories[name] = self 

221 

222 if parent._root is not None: 222 ↛ 225line 222 didn't jump to line 225 because the condition on line 222 was always true

223 self._root = parent._root 

224 

225 if collectSubdirectories: 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true

226 self._collectSubdirectories() 

227 

228 def _collectSubdirectories(self) -> None: 

229 with Stopwatch() as sw1: 

230 self._scanSubdirectories() 

231 

232 with Stopwatch() as sw2: 

233 self._aggregateSizes() 

234 

235 self._scanDuration = sw1.Duration 

236 self._aggregateDuration = sw2.Duration 

237 

238 def _scanSubdirectories(self) -> None: 

239 try: 

240 items = scandir(directoryPath := self.Path) 

241 except PermissionError as ex: 

242 return 

243 

244 for dirEntry in items: 

245 if dirEntry.is_dir(follow_symlinks=False): 

246 subdirectory = Directory(dirEntry.name, collectSubdirectories=True, parent=self) 

247 elif dirEntry.is_file(follow_symlinks=False): 

248 id = dirEntry.inode() 

249 if id in self._root._ids: 

250 file = self._root._ids[id] 

251 

252 hardLink = Filename(dirEntry.name, file=file, parent=self) 

253 else: 

254 s = dirEntry.stat(follow_symlinks=False) 

255 filename = Filename(dirEntry.name, parent=self) 

256 file = File(id, s.st_size, parent=filename) 

257 

258 self._root._ids[id] = file 

259 elif dirEntry.is_symlink(): 

260 target = Path(readlink(directoryPath / dirEntry.name)) 

261 symlink = SymbolicLink(dirEntry.name, target, parent=self) 

262 else: 

263 raise FilesystemException(f"Unknown directory element.") 

264 

265 def _connectSymbolicLinks(self) -> None: 

266 for dir in self._subdirectories.values(): 

267 dir._connectSymbolicLinks() 

268 

269 for link in self._symbolicLinks.values(): 

270 if link._target.is_absolute(): 

271 pass 

272 else: 

273 target = self 

274 for elem in link._target.parts: 

275 if elem == ".": 

276 continue 

277 elif elem == "..": 

278 target = target._parent 

279 continue 

280 

281 try: 

282 target = target._subdirectories[elem] 

283 continue 

284 except KeyError: 

285 pass 

286 

287 try: 

288 target = target._files[elem] 

289 continue 

290 except KeyError: 

291 pass 

292 

293 try: 

294 target = target._symbolicLinks[elem] 

295 continue 

296 except KeyError: 

297 pass 

298 

299 target.AddLinkSources(link) 

300 

301 def _aggregateSizes(self) -> None: 

302 self._size = ( 

303 sum(dir._size for dir in self._subdirectories.values()) + 

304 sum(file._file._size for file in self._files.values()) 

305 ) 

306 

307 @Element.Root.setter 

308 def Root(self, value: "Root") -> None: 

309 Element.Root.fset(self, value) 

310 

311 for subdir in self._subdirectories.values(): 311 ↛ 312line 311 didn't jump to line 312 because the loop on line 311 never started

312 subdir.Root = value 

313 

314 for file in self._files.values(): 

315 file.Root = value 

316 

317 for link in self._symbolicLinks.values(): 317 ↛ 318line 317 didn't jump to line 318 because the loop on line 317 never started

318 link.Root = value 

319 

320 @Element.Parent.setter 

321 def Parent(self, value: _ParentType) -> None: 

322 Element.Parent.fset(self, value) 

323 

324 value._subdirectories[self._name] = self 

325 

326 if isinstance(value, Root): 326 ↛ exitline 326 didn't return from function 'Parent' because the condition on line 326 was always true

327 self.Root = value 

328 

329 @readonly 

330 def Count(self) -> int: 

331 """ 

332 Read-only property to access the number of elements in a directory. 

333 

334 :returns: Number of files plus subdirectories. 

335 """ 

336 return len(self._subdirectories) + len(self._files) + len(self._symbolicLinks) 

337 

338 @readonly 

339 def FileCount(self) -> int: 

340 """ 

341 Read-only property to access the number of files in a directory. 

342 

343 .. hint:: 

344 

345 Files include regular files and symbolic links. 

346 

347 :returns: Number of files. 

348 """ 

349 return len(self._files) + len(self._symbolicLinks) 

350 

351 @readonly 

352 def RegularFileCount(self) -> int: 

353 """ 

354 Read-only property to access the number of regular files in a directory. 

355 

356 :returns: Number of regular files. 

357 """ 

358 return len(self._files) 

359 

360 @readonly 

361 def SymbolicLinkCount(self) -> int: 

362 """ 

363 Read-only property to access the number of symbolic links in a directory. 

364 

365 :returns: Number of symbolic links. 

366 """ 

367 return len(self._symbolicLinks) 

368 

369 @readonly 

370 def SubdirectoryCount(self) -> int: 

371 """ 

372 Read-only property to access the number of subdirectories in a directory. 

373 

374 :returns: Number of subdirectories. 

375 """ 

376 return len(self._subdirectories) 

377 

378 @readonly 

379 def TotalFileCount(self) -> int: 

380 """ 

381 Read-only property to access the total number of files in all child hierarchy levels (recursively). 

382 

383 .. hint:: 

384 

385 Files include regular files and symbolic links. 

386 

387 :returns: Total number of files. 

388 """ 

389 return sum(d.TotalFileCount for d in self._subdirectories.values()) + len(self._files) + len(self._symbolicLinks) 

390 

391 @readonly 

392 def TotalRegularFileCount(self) -> int: 

393 """ 

394 Read-only property to access the total number of regular files in all child hierarchy levels (recursively). 

395 

396 :returns: Total number of regular files. 

397 """ 

398 return sum(d.TotalRegularFileCount for d in self._subdirectories.values()) + len(self._files) 

399 

400 @readonly 

401 def TotalSymbolicLinkCount(self) -> int: 

402 """ 

403 Read-only property to access the total number of symbolic links in all child hierarchy levels (recursively). 

404 

405 :returns: Total number of symbolic links. 

406 """ 

407 return sum(d.TotalSymbolicLinkCount for d in self._subdirectories.values()) + len(self._symbolicLinks) 

408 

409 @readonly 

410 def TotalSubdirectoryCount(self) -> int: 

411 """ 

412 Read-only property to access the total number of subdirectories in all child hierarchy levels (recursively). 

413 

414 :returns: Total number of subdirectories. 

415 """ 

416 return len(self._subdirectories) + sum(d.TotalSubdirectoryCount for d in self._subdirectories.values()) 

417 

418 @readonly 

419 def Subdirectories(self) -> Generator["Directory", None, None]: 

420 """ 

421 Iterate all direct subdirectories of the directory. 

422 

423 :returns: A generator to iterate all direct subdirectories. 

424 """ 

425 return (d for d in self._subdirectories.values()) 

426 

427 @readonly 

428 def Files(self) -> Generator[Union["Filename", "SymbolicLink"], None, None]: 

429 """ 

430 Iterate all direct files of the directory. 

431 

432 .. hint:: 

433 

434 Files include regular files and symbolic links. 

435 

436 :returns: A generator to iterate all direct files. 

437 """ 

438 return (f for f in chain(self._files.values(), self._symbolicLinks.values())) 

439 

440 @readonly 

441 def RegularFiles(self) -> Generator["Filename", None, None]: 

442 """ 

443 Iterate all direct regular files of the directory. 

444 

445 :returns: A generator to iterate all direct regular files. 

446 """ 

447 return (f for f in self._files.values()) 

448 

449 @readonly 

450 def SymbolicLinks(self) -> Generator["SymbolicLink", None, None]: 

451 """ 

452 Iterate all direct symbolic links of the directory. 

453 

454 :returns: A generator to iterate all direct symbolic links. 

455 """ 

456 return (l for l in self._symbolicLinks.values()) 

457 

458 @readonly 

459 def Path(self) -> Path: 

460 """ 

461 Read-only property to access the equivalent Path instance for accessing the represented directory. 

462 

463 :returns: Path to the directory. 

464 :raises FilesystemException: If no parent is set. 

465 """ 

466 if self._path is not None: 

467 return self._path 

468 

469 if self._parent is None: 

470 raise FilesystemException(f"No parent or root set for directory.") 

471 

472 self._path = self._parent.Path / self._name 

473 return self._path 

474 

475 @readonly 

476 def ScanDuration(self) -> float: 

477 """ 

478 Read-only property to access the time needed to scan a directory structure including all subelements (recursively). 

479 

480 :returns: The scan duration in seconds. 

481 :raises FilesystemException: If the directory was not scanned. 

482 """ 

483 if self._scanDuration is None: 

484 raise FilesystemException(f"Directory was not scanned, yet.") 

485 

486 return self._scanDuration 

487 

488 @readonly 

489 def AggregateDuration(self) -> float: 

490 """ 

491 Read-only property to access the time needed to aggregate the directory's and subelement's properties (recursively). 

492 

493 :returns: The aggregation duration in seconds. 

494 :raises FilesystemException: If the directory properties were not aggregated. 

495 """ 

496 if self._scanDuration is None: 

497 raise FilesystemException(f"Directory properties were not aggregated, yet.") 

498 

499 return self._aggregateDuration 

500 

501 def Copy(self, parent: Nullable["Directory"] = None) -> "Directory": 

502 """ 

503 Copy the directory structure including all subelements and link it to the given parent. 

504 

505 .. hint:: 

506 

507 Statistics like aggregated directory size are copied too. |br| 

508 There is no rescan or repeated aggregation needed. 

509 

510 :param parent: The parent element of the copied directory. 

511 :returns: A deep copy of the directory structure. 

512 """ 

513 dir = Directory(self._name, parent=parent) 

514 dir._size = self._size 

515 

516 for subdir in self._subdirectories.values(): 

517 subdir.Copy(dir) 

518 

519 for file in self._files.values(): 

520 file.Copy(dir) 

521 

522 for link in self._symbolicLinks.values(): 

523 link.Copy(dir) 

524 

525 return dir 

526 

527 def Collapse(self, func: Callable[["Directory"], bool]) -> bool: 

528 # if len(self._subdirectories) == 0 or all(subdir.Collapse(func) for subdir in self._subdirectories.values()): 

529 if len(self._subdirectories) == 0: 

530 if func(self): 

531 # print(f"collapse 1 {self.Path}") 

532 self._collapsed = True 

533 self._subdirectories.clear() 

534 self._files.clear() 

535 self._symbolicLinks.clear() 

536 

537 return True 

538 else: 

539 return False 

540 

541 # if all(subdir.Collapse(func) for subdir in self._subdirectories.values()) 

542 collapsible = True 

543 for subdir in self._subdirectories.values(): 

544 result = subdir.Collapse(func) 

545 collapsible = collapsible and result 

546 

547 if collapsible: 

548 # print(f"collapse 2 {self.Path}") 

549 self._collapsed = True 

550 self._subdirectories.clear() 

551 self._files.clear() 

552 self._symbolicLinks.clear() 

553 

554 return True 

555 else: 

556 return False 

557 

558 def ToTree(self, format: Nullable[Callable[[Node], str]] = None) -> Node: 

559 """ 

560 Convert the directory to a :class:`~pyTooling.Tree.Node`. 

561 

562 The node's :attr:`~pyTooling.Tree.Node.Value` field contains a reference to the directory. Additional data is 

563 attached to the node's key-value store: 

564 

565 ``kind`` 

566 The node's kind. See :class:`NodeKind`. 

567 ``size`` 

568 The directory's aggregated size. 

569 

570 :param format: A user defined formatting function for tree nodes. 

571 :returns: A tree node representing this directory. 

572 """ 

573 if format is None: 

574 def format(node: Node) -> str: 

575 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}" 

576 

577 directoryNode = Node( 

578 value=self, 

579 keyValuePairs={ 

580 "kind": NodeKind.File, 

581 "size": self._size 

582 }, 

583 format=format 

584 ) 

585 directoryNode.AddChildren( 

586 e.ToTree(format) for e in chain(self._subdirectories.values()) #, self._files.values(), self._symbolicLinks.values()) 

587 ) 

588 

589 return directoryNode 

590 

591 def __eq__(self, other) -> bool: 

592 """ 

593 Compare two Directory instances for equality. 

594 

595 :param other: Parameter to compare against. 

596 :returns: ``True``, if both directories and all its subelements are equal. 

597 :raises TypeError: If parameter ``other`` is not of type :class:`Directory`. 

598 """ 

599 if not isinstance(other, Directory): 

600 ex = TypeError("Parameter 'other' is not of type Directory.") 

601 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.") 

602 raise ex 

603 

604 if not all(dir1 == dir2 for _, dir1, dir2 in zipdicts(self._subdirectories, other._subdirectories)): 

605 return False 

606 

607 if not all(file1 == file2 for _, file1, file2 in zipdicts(self._files, other._files)): 

608 return False 

609 

610 if not all(link1 == link2 for _, link1, link2 in zipdicts(self._symbolicLinks, other._symbolicLinks)): 

611 return False 

612 

613 return True 

614 

615 def __ne__(self, other: Any) -> bool: 

616 """ 

617 Compare two Directory instances for inequality. 

618 

619 :param other: Parameter to compare against. 

620 :returns: ``True``, if both directories and all its subelements are unequal. 

621 :raises TypeError: If parameter ``other`` is not of type :class:`Directory`. 

622 """ 

623 return not self.__eq__(other) 

624 

625 def __repr__(self) -> str: 

626 return f"Directory: {self.Path}" 

627 

628 def __str__(self) -> str: 

629 return self._name 

630 

631 

632@export 

633class Filename(Element[Directory]): 

634 _file: Nullable["File"] 

635 

636 def __init__( 

637 self, 

638 name: str, 

639 file: Nullable["File"] = None, 

640 parent: Nullable[Directory] = None 

641 ) -> None: 

642 super().__init__(name, None, parent) 

643 

644 if file is None: 644 ↛ 647line 644 didn't jump to line 647 because the condition on line 644 was always true

645 self._file = None 

646 else: 

647 self._file = file 

648 file._parents.append(self) 

649 

650 if parent is not None: 

651 parent._files[name] = self 

652 

653 if parent._root is not None: 

654 self._root = parent._root 

655 

656 @Element.Root.setter 

657 def Root(self, value: "Root") -> None: 

658 self._root = value 

659 

660 if self._file is not None: 660 ↛ exitline 660 didn't return from function 'Root' because the condition on line 660 was always true

661 self._file._root = value 

662 

663 @Element.Parent.setter 

664 def Parent(self, value: _ParentType) -> None: 

665 Element.Parent.fset(self, value) 

666 

667 value._files[self._name] = self 

668 

669 if isinstance(value, Root): 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true

670 self.Root = value 

671 

672 @readonly 

673 def File(self) -> Nullable["File"]: 

674 return self._file 

675 

676 @readonly 

677 def Size(self) -> int: 

678 if self._file is None: 

679 raise ToolingException(f"Filename isn't linked to a File object.") 

680 

681 return self._file._size 

682 

683 @readonly 

684 def Path(self) -> Path: 

685 if self._parent is None: 

686 raise ToolingException(f"Filename has no parent object.") 

687 

688 return self._parent.Path / self._name 

689 

690 def Copy(self, parent: Directory) -> "Filename": 

691 fileID = self._file._id 

692 

693 if fileID in parent._root._ids: 

694 file = parent._root._ids[fileID] 

695 else: 

696 fileSize = self._file._size 

697 file = File(fileID, fileSize) 

698 

699 parent._root._ids[fileID] = file 

700 

701 return Filename(self._name, file, parent=parent) 

702 

703 def ToTree(self) -> Node: 

704 def format(node: Node) -> str: 

705 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}" 

706 

707 fileNode = Node( 

708 value=self, 

709 keyValuePairs={ 

710 "kind": NodeKind.File, 

711 "size": self._size 

712 }, 

713 format=format 

714 ) 

715 

716 return fileNode 

717 

718 def __eq__(self, other) -> bool: 

719 """ 

720 Compare two Filename instances for equality. 

721 

722 :param other: Parameter to compare against. 

723 :returns: ``True``, if both filenames are equal. 

724 :raises TypeError: If parameter ``other`` is not of type :class:`Filename`. 

725 """ 

726 if not isinstance(other, Filename): 

727 ex = TypeError("Parameter 'other' is not of type Filename.") 

728 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.") 

729 raise ex 

730 

731 return self._name == other._name and self.Size == other.Size 

732 

733 def __ne__(self, other: Any) -> bool: 

734 """ 

735 Compare two Filename instances for inequality. 

736 

737 :param other: Parameter to compare against. 

738 :returns: ``True``, if both filenames are unequal. 

739 :raises TypeError: If parameter ``other`` is not of type :class:`Filename`. 

740 """ 

741 if not isinstance(other, Filename): 

742 ex = TypeError("Parameter 'other' is not of type Filename.") 

743 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.") 

744 raise ex 

745 

746 return self._name != other._name or self.Size != other.Size 

747 

748 def __repr__(self) -> str: 

749 return f"File: {self.Path}" 

750 

751 def __str__(self) -> str: 

752 return self._name 

753 

754 

755@export 

756class SymbolicLink(Element[Directory]): 

757 _target: Path 

758 

759 def __init__( 

760 self, 

761 name: str, 

762 target: Path, 

763 parent: Nullable[Directory] 

764 ) -> None: 

765 super().__init__(name, None, parent) 

766 

767 self._target = target 

768 

769 if parent is not None: 

770 parent._symbolicLinks[name] = self 

771 

772 if parent._root is not None: 

773 self._root = parent._root 

774 

775 @readonly 

776 def Path(self) -> Path: 

777 return self._parent.Path / self._name 

778 

779 @readonly 

780 def Target(self) -> Path: 

781 return self._target 

782 

783 def Copy(self, parent: Directory) -> "SymbolicLink": 

784 return SymbolicLink(self._name, self._target, parent=parent) 

785 

786 def ToTree(self) -> Node: 

787 def format(node: Node) -> str: 

788 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}" 

789 

790 symbolicLinkNode = Node( 

791 value=self, 

792 keyValuePairs={ 

793 "kind": NodeKind.SymbolicLink, 

794 "size": self._size 

795 }, 

796 format=format 

797 ) 

798 

799 return symbolicLinkNode 

800 

801 def __eq__(self, other) -> bool: 

802 """ 

803 Compare two SymbolicLink instances for equality. 

804 

805 :param other: Parameter to compare against. 

806 :returns: ``True``, if both symbolic links are equal. 

807 :raises TypeError: If parameter ``other`` is not of type :class:`SymbolicLink`. 

808 """ 

809 if not isinstance(other, SymbolicLink): 

810 ex = TypeError("Parameter 'other' is not of type SymbolicLink.") 

811 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.") 

812 raise ex 

813 

814 return self._name == other._name and self._target == other._target 

815 

816 def __ne__(self, other: Any) -> bool: 

817 """ 

818 Compare two SymbolicLink instances for inequality. 

819 

820 :param other: Parameter to compare against. 

821 :returns: ``True``, if both symbolic links are unequal. 

822 :raises TypeError: If parameter ``other`` is not of type :class:`SymbolicLink`. 

823 """ 

824 if not isinstance(other, SymbolicLink): 

825 ex = TypeError("Parameter 'other' is not of type SymbolicLink.") 

826 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.") 

827 raise ex 

828 

829 return self._name != other._name or self._target != other._target 

830 

831 def __repr__(self) -> str: 

832 return f"SymLink: {self.Path} -> {self._target}" 

833 

834 def __str__(self) -> str: 

835 return self._name 

836 

837 

838@export 

839class Root(Directory): 

840 _ids: Dict[int, "File"] #: Dictionary of file identifier - file objects pairs found while scanning the directory structure. 

841 

842 def __init__( 

843 self, 

844 rootDirectory: Path, 

845 collectSubdirectories: bool = True 

846 ) -> None: 

847 if rootDirectory is None: 847 ↛ 848line 847 didn't jump to line 848 because the condition on line 847 was never true

848 raise ValueError(f"Parameter 'path' is None.") 

849 elif not isinstance(rootDirectory, Path): 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true

850 raise TypeError(f"Parameter 'path' is not of type Path.") 

851 elif not rootDirectory.exists(): 851 ↛ 852line 851 didn't jump to line 852 because the condition on line 851 was never true

852 raise ToolingException(f"Path '{rootDirectory}' doesn't exist.") from FileNotFoundError(rootDirectory) 

853 

854 self._ids = {} 

855 

856 super().__init__(rootDirectory.name) 

857 self._root = self 

858 self._path = rootDirectory 

859 

860 if collectSubdirectories: 860 ↛ 861line 860 didn't jump to line 861 because the condition on line 860 was never true

861 self._collectSubdirectories() 

862 self._connectSymbolicLinks() 

863 

864 @readonly 

865 def TotalHardLinkCount(self) -> int: 

866 return sum(l for f in self._ids.values() if (l := len(f._parents)) > 1) 

867 

868 @readonly 

869 def TotalHardLinkCount2(self) -> int: 

870 return sum(1 for f in self._ids.values() if len(f._parents) > 1) 

871 

872 @readonly 

873 def TotalHardLinkCount3(self) -> int: 

874 return sum(1 for f in self._ids.values() if len(f._parents) == 1) 

875 

876 @readonly 

877 def Size2(self) -> int: 

878 return sum(f._size for f in self._ids.values() if len(f._parents) > 1) 

879 

880 @readonly 

881 def Size3(self) -> int: 

882 return sum(f._size * len(f._parents) for f in self._ids.values() if len(f._parents) > 1) 

883 

884 @readonly 

885 def TotalUniqueFileCount(self) -> int: 

886 return len(self._ids) 

887 

888 @readonly 

889 def Path(self) -> Path: 

890 """ 

891 Read-only property to access the path of the filesystem statistics root. 

892 

893 :returns: Path to the root of the filesystem statistics root directory. 

894 """ 

895 return self._path 

896 

897 def Copy(self) -> "Root": 

898 """ 

899 Copy the directory structure including all subelements and link it to the given parent. 

900 

901 The duration for the deep copy process is provided in :attr:`ScanDuration` 

902 

903 .. hint:: 

904 

905 Statistics like aggregated directory size are copied too. |br| 

906 There is no rescan or repeated aggregation needed. 

907 

908 :returns: A deep copy of the directory structure. 

909 """ 

910 with Stopwatch() as sw: 

911 root = Root(self._path, False) 

912 root._size = self._size 

913 

914 for subdir in self._subdirectories.values(): 

915 subdir.Copy(root) 

916 

917 for file in self._files.values(): 

918 file.Copy(root) 

919 

920 for link in self._symbolicLinks.values(): 

921 link.Copy(root) 

922 

923 root._scanDuration = sw.Duration 

924 root._aggregateDuration = 0.0 

925 

926 return root 

927 

928 def __repr__(self) -> str: 

929 return f"Root: {self.Path} (dirs: {self.TotalSubdirectoryCount}, files: {self.TotalRegularFileCount}, symlinks: {self.TotalSymbolicLinkCount})" 

930 

931 def __str__(self) -> str: 

932 return self._name 

933 

934 

935@export 

936class File(Base): 

937 _id: int 

938 _parents: List[Filename] 

939 

940 def __init__( 

941 self, 

942 id: int, 

943 size: int, 

944 parent: Nullable[Filename] = None 

945 ) -> None: 

946 self._id = id 

947 if parent is None: 

948 super().__init__(None, size) 

949 self._parents = [] 

950 else: 

951 super().__init__(parent._root, size) 

952 self._parents = [parent] 

953 parent._file = self 

954 

955 @readonly 

956 def ID(self) -> int: 

957 """ 

958 Read-only property to access the file object's unique identifier. 

959 

960 :returns: Unique file object identifier. 

961 """ 

962 return self._id 

963 

964 @readonly 

965 def Parents(self) -> List[Filename]: 

966 """ 

967 Read-only property to access the list of filenames using the file object. 

968 

969 .. hint:: 

970 

971 This allows to check if a file object has multiple filenames a.k.a hardlinks. 

972 

973 :returns: List of filenames for the file object. 

974 """ 

975 return self._parents 

976 

977 def AddParent(self, file: Filename) -> None: 

978 if file._file is not None: 978 ↛ 979line 978 didn't jump to line 979 because the condition on line 978 was never true

979 raise ToolingException(f"Filename is already referencing an other file object ({file._file._id}).") 

980 

981 self._parents.append(file) 

982 file._file = self 

983 

984 if file._root is not None: 

985 self._root = file._root