Coverage for pyTooling/Filesystem/__init__.py: 49%
577 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-07 22:39 +0000
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"""
32An object-oriented file system abstraction for directory, file, symbolic link, ... statistics collection.
34.. important::
36 This isn't a replacement of :mod:`pathlib` introduced with Python 3.4.
37"""
38from os import scandir, readlink
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, Iterator, Set
45from pyTooling.Decorators import readonly, export
46from pyTooling.Exceptions import ToolingException
47from pyTooling.MetaClasses import ExtendedType
48from pyTooling.Common import getFullyQualifiedName, zipdicts
49from pyTooling.Warning import WarningCollector, Warning
50from pyTooling.Stopwatch import Stopwatch
51from pyTooling.Tree import Node
54__all__ = ["_ParentType"]
57_ParentType = TypeVar("_ParentType", bound="Element")
58"""The type variable for a parent reference."""
61@export
62class FilesystemException(ToolingException):
63 """Base-exception of all exceptions raised by :mod:`pyTooling.Filesystem`."""
66@export
67class PermissionWarning(Warning):
68 _path: Path
70 def __init__(self, path: Path, *args) -> None:
71 super().__init__(*args)
72 self._path = path
74 @readonly
75 def Path(self) -> Path:
76 """
77 Read-only property to access the path that couldn't be read (:attr:`_path`).
79 :returns: The path that raised a :exc:`PermissionError`.
80 """
81 return self._path
84@export
85class NodeKind(Enum):
86 """
87 Node kind for filesystem elements in a :ref:`tree <STRUCT/Tree>`.
89 This enumeration is used when converting the filesystem statistics tree to an instance of :mod:`pyTooling.Tree`.
90 """
91 Directory = 0 #: Node represents a directory.
92 File = 1 #: Node represents a regular file.
93 SymbolicLink = 2 #: Node represents a symbolic link.
96@export
97class Base(metaclass=ExtendedType, slots=True):
98 """
99 Base-class for all filesystem elements in :mod:`pyTooling.Filesystem`.
101 It implements a size and a reference to the root element of the filesystem.
102 """
103 _root: Nullable["Root"] #: Reference to the root of the filesystem statistics scope.
104 _size: Nullable[int] #: Actual or aggregated size of the filesystem element.
106 def __init__(
107 self,
108 size: Nullable[int],
109 root: Nullable["Root"]
110 ) -> None:
111 """
112 Initialize the base-class with filesystem element size and root reference.
114 :param size: Optional size of the element.
115 :param root: Optional reference to the filesystem root element.
116 """
117 if size is not None and not isinstance(size, int): 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 ex = TypeError("Parameter 'size' is not of type 'int'.")
119 ex.add_note(f"Got type '{getFullyQualifiedName(size)}'.")
120 raise ex
122 if root is not None and not isinstance(root, Root): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 ex = TypeError("Parameter 'root' is not of type 'Root'.")
124 ex.add_note(f"Got type '{getFullyQualifiedName(root)}'.")
125 raise ex
127 self._size = size
128 self._root = root
130 @property
131 def Root(self) -> Nullable["Root"]:
132 """
133 Property to access the root of the filesystem statistics scope.
135 :returns: Root of the filesystem statistics scope.
136 """
137 return self._root
139 @Root.setter
140 def Root(self, value: "Root") -> None:
141 if value is None: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 raise ValueError(f"Parameter 'value' is None.")
143 elif not isinstance(value, Root): 143 ↛ 144line 143 didn't jump to line 144 because the condition on line 143 was never true
144 ex = TypeError("Parameter 'value' is not of type 'Root'.")
145 ex.add_note(f"Got type '{getFullyQualifiedName(value)}'.")
146 raise ex
148 self._root = value
150 @readonly
151 def Size(self) -> int:
152 """
153 Read-only property to access the element's size in Bytes.
155 :returns: Size in Bytes.
156 :raises FilesystemException: If size is not computed, yet.
157 """
158 if self._size is None: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 raise FilesystemException("Size is not computed, yet.")
161 return self._size
163 # FIXME: @abstractmethod
164 def ToTree(self) -> Node:
165 """
166 Convert a filesystem element to a node in :mod:`pyTooling.Tree`.
168 The node's :attr:`~pyTooling.Tree.Node.Value` field contains a reference to the filesystem element. Additional data
169 will be stored in the node's key-value store.
171 :returns: A tree's node referencing this filesystem element.
172 """
173 raise NotImplementedError()
176@export
177class Element(Base, Generic[_ParentType]):
178 """
179 Base-class for all named elements within a filesystem.
181 It adds a name, parent reference and list of symbolic-link sources.
183 .. hint::
185 Symbolic link sources are reverse references describing which symbolic links point to this element.
186 """
187 _name: str #: Name of the filesystem element.
188 _parent: _ParentType #: Reference to the filesystem element's parent (:class:`Directory`)
189 _linkSources: List["SymbolicLink"] #: A list of symbolic links pointing to this filesystem element.
191 def __init__(
192 self,
193 name: str,
194 size: Nullable[int] = None,
195 parent: Nullable[_ParentType] = None
196 ) -> None:
197 """
198 Initialize the element base-class with name, size and parent reference.
200 :param name: Name of the element.
201 :param size: Optional size of the element.
202 :param parent: Optional parent reference.
203 """
204 if name is None: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true
205 raise ValueError(f"Parameter 'name' is None.")
206 elif not isinstance(name, str): 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 ex = TypeError("Parameter 'name' is not of type 'str'.")
208 ex.add_note(f"Got type '{getFullyQualifiedName(name)}'.")
209 raise ex
211 self._name = name
213 if parent is None:
214 super().__init__(size, None)
215 self._parent = None
216 else:
217 if not isinstance(parent, Directory): 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 ex = TypeError("Parameter 'parent' is not of type 'Directory'.")
219 ex.add_note(f"Got type '{getFullyQualifiedName(parent)}'.")
220 raise ex
222 super().__init__(size, parent._root)
223 self._parent = parent
225 self._linkSources = []
227 @property
228 def Parent(self) -> _ParentType:
229 """
230 Property to access the element's parent.
232 :returns: Parent element.
233 """
234 return self._parent
236 @Parent.setter
237 def Parent(self, value: _ParentType) -> None:
238 if value is None: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 raise ValueError(f"Parameter 'value' is None.")
240 elif not isinstance(value, Directory): 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true
241 ex = TypeError("Parameter 'value' is not of type 'Directory'.")
242 ex.add_note(f"Got type '{getFullyQualifiedName(value)}'.")
243 raise ex
245 self._parent = value
247 if value._root is not None:
248 self._root = value._root
250 @readonly
251 def Name(self) -> str:
252 """
253 Read-only property to access the element's name.
255 :returns: Element name.
256 """
257 return self._name
259 @readonly
260 def Path(self) -> Path:
261 """
262 Read-only property to access the element's path.
264 :returns: Path of the element.
265 """
266 raise NotImplementedError(f"Property 'Path' is abstract.")
268 @readonly
269 def LinkSources(self) -> List["SymbolicLink"]:
270 """
271 Read-only property to access the symbolic links pointing to this element (:attr:`_linkSources`).
273 :returns: List of symbolic links targeting this element.
274 """
275 return self._linkSources
277 def AddLinkSources(self, source: "SymbolicLink") -> None:
278 """
279 Add a link source of a symbolic link to the named element (reverse reference).
281 :param source: The referenced symbolic link.
282 """
283 if not isinstance(source, SymbolicLink):
284 ex = TypeError("Parameter 'source' is not of type 'SymbolicLink'.")
285 ex.add_note(f"Got type '{getFullyQualifiedName(source)}'.")
286 raise ex
288 source._isConnected = True
289 source._isBroken = False
290 source._isOutOfRange = False
291 self._linkSources.append(source)
294@export
295class Directory(Element["Directory"]):
296 """
297 A **directory** represents a directory in the filesystem, which contains subdirectories, regular files and symbolic links.
299 While scanning for subelements, the directory is populated with elements. Every file object added, gets registered in
300 the filesystems :class:`Root` for deduplication. In case a file identifier already exists, the found filename will
301 reference the same file objects. In turn, the file objects has then references to multiple filenames (parents). This
302 allows to detect :term:`hardlinks <hardlink>`.
304 The time needed for scanning the directory and its subelements is provided via :data:`ScanDuration`.
306 After scnaning the directory for subelements, certain directory properties get aggregated. The time needed for
307 aggregation is provided via :data:`AggregateDuration`.
308 """
310 _path: Nullable[Path] #: Cached :class:`~pathlib.Path` object of this directory.
311 _subdirectories: Dict[str, "Directory"] #: Dictionary containing name-:class:`Directory` pairs.
312 _files: Dict[str, "Filename"] #: Dictionary containing name-:class:`Filename` pairs.
313 _symbolicLinks: Dict[str, "SymbolicLink"] #: Dictionary containing name-:class:`SymbolicLink` pairs.
314 _filesSize: int #: Aggregated size of all direct files.
315 _collapsed: bool #: True, if this directory was collapsed. It contains no subelements.
316 _scanDuration: Nullable[float] #: Duration for scanning the directory and all its subelements.
317 _aggregateDuration: Nullable[float] #: Duration for aggregating all subelements.
319 def __init__(
320 self,
321 name: str,
322 collectSubdirectories: bool = False,
323 parent: Nullable["Directory"] = None
324 ) -> None:
325 """
326 Initialize the directory with name and parent reference.
328 :param name: Name of the element.
329 :param collectSubdirectories: If true, collect subdirectory statistics.
330 :param parent: Optional parent reference.
331 """
332 super().__init__(name, None, parent)
334 self._path = None
335 self._subdirectories = {}
336 self._files = {}
337 self._symbolicLinks = {}
338 self._filesSize = 0
339 self._collapsed = False
340 self._scanDuration = None
341 self._aggregateDuration = None
343 if parent is not None:
344 parent._subdirectories[name] = self
346 if parent._root is not None: 346 ↛ 349line 346 didn't jump to line 349 because the condition on line 346 was always true
347 self._root = parent._root
349 if collectSubdirectories: 349 ↛ 350line 349 didn't jump to line 350 because the condition on line 349 was never true
350 self.CollectSubdirectories()
352 def CollectSubdirectories(self) -> None:
353 """
354 Helper method for scanning subdirectories and aggregating found element sizes therein.
355 """
356 self.ScanSubdirectories()
357 self.AggregateSizes()
359 def ScanSubdirectories(self) -> None:
360 """
361 Helper method for scanning subdirectories (recursively) and building a
362 :class:`Directory`-:class:`Filename`-:class:`File` object tree.
364 If a file refers to the same filesystem internal unique ID, a hardlink (two or more filenames) to the same file
365 storage object is assumed.
366 """
367 with Stopwatch() as sw1:
368 try:
369 items = scandir(directoryPath := self.Path)
370 except PermissionError as ex:
371 return WarningCollector.Raise(PermissionWarning(self.Path), ex)
373 for dirEntry in items:
374 if dirEntry.is_dir(follow_symlinks=False):
375 _ = Directory(dirEntry.name, collectSubdirectories=True, parent=self)
376 elif dirEntry.is_file(follow_symlinks=False):
377 id = dirEntry.inode()
378 if id in self._root._ids:
379 file = self._root._ids[id]
381 _ = Filename(dirEntry.name, file=file, parent=self)
382 else:
383 s = dirEntry.stat(follow_symlinks=False)
384 filename = Filename(dirEntry.name, parent=self)
385 file = File(id, s.st_size, parent=filename)
387 self._root._ids[id] = file
388 elif dirEntry.is_symlink():
389 target = Path(readlink(directoryPath / dirEntry.name))
390 _ = SymbolicLink(dirEntry.name, target, parent=self)
391 else:
392 raise FilesystemException(f"Unknown directory element.")
394 self._scanDuration = sw1.Duration
396 def ResolveSymbolicLinks(self) -> None:
397 for dir in self._subdirectories.values():
398 dir.ResolveSymbolicLinks()
400 for link in self._symbolicLinks.values(): 400 ↛ 401line 400 didn't jump to line 401 because the loop on line 400 never started
401 if link._target.is_absolute():
402 # todo: resolve path and check if target is in range, otherwise add to out-of-range list
403 pass
404 else:
405 target = self
406 for elem in link._target.parts:
407 if elem == ".":
408 continue
409 elif elem == "..":
410 if (target := target._parent) is None:
411 self._root.RegisterUnconnectedSymbolicLink(link)
412 break
414 continue
416 try:
417 target = target._subdirectories[elem]
418 continue
419 except KeyError:
420 pass
422 try:
423 target = target._files[elem]
424 continue
425 except KeyError:
426 pass
428 try:
429 target = target._symbolicLinks[elem]
430 continue
431 except KeyError:
432 self._root.RegisterBrokenSymbolicLink(link)
433 break
434 else:
435 target.AddLinkSources(link)
437 def AggregateSizes(self) -> Set["File"]:
438 with Stopwatch() as sw2:
439 aggregatedFiles = set()
441 self._size = 0
442 self._filesSize = 0
443 for dir in self._subdirectories.values():
444 aggregatedFiles |= dir.AggregateSizes()
445 self._size += dir._size
447 for filename in self._files.values():
448 if (file := filename._file) not in aggregatedFiles: 448 ↛ 447line 448 didn't jump to line 447 because the condition on line 448 was always true
449 self._filesSize += file._size
450 aggregatedFiles.add(file)
452 self._size += self._filesSize
454 self._aggregateDuration = sw2.Duration
456 return aggregatedFiles
458 @Element.Root.setter
459 def Root(self, value: "Root") -> None:
460 Element.Root.fset(self, value)
462 for subdir in self._subdirectories.values(): 462 ↛ 463line 462 didn't jump to line 463 because the loop on line 462 never started
463 subdir.Root = value
465 for file in self._files.values():
466 file.Root = value
468 for link in self._symbolicLinks.values(): 468 ↛ 469line 468 didn't jump to line 469 because the loop on line 468 never started
469 link.Root = value
471 @Element.Parent.setter
472 def Parent(self, value: _ParentType) -> None:
473 Element.Parent.fset(self, value)
475 value._subdirectories[self._name] = self
477 if isinstance(value, Root): 477 ↛ exitline 477 didn't return from function 'Parent' because the condition on line 477 was always true
478 self.Root = value
480 @readonly
481 def Count(self) -> int:
482 """
483 Read-only property to return the number of elements in a directory.
485 :returns: Number of files plus subdirectories.
486 """
487 return len(self._subdirectories) + len(self._files) + len(self._symbolicLinks)
489 @readonly
490 def FileCount(self) -> int:
491 """
492 Read-only property to return the number of files in a directory.
494 .. hint::
496 Files include regular files and symbolic links.
498 :returns: Number of files.
499 """
500 return len(self._files) + len(self._symbolicLinks)
502 @readonly
503 def RegularFileCount(self) -> int:
504 """
505 Read-only property to return the number of regular files in a directory.
507 :returns: Number of regular files.
508 """
509 return len(self._files)
511 @readonly
512 def SymbolicLinkCount(self) -> int:
513 """
514 Read-only property to return the number of symbolic links in a directory.
516 :returns: Number of symbolic links.
517 """
518 return len(self._symbolicLinks)
520 @readonly
521 def SubdirectoryCount(self) -> int:
522 """
523 Read-only property to return the number of subdirectories in a directory.
525 :returns: Number of subdirectories.
526 """
527 return len(self._subdirectories)
529 @readonly
530 def TotalFileCount(self) -> int:
531 """
532 Read-only property to return the total number of files in all child hierarchy levels (recursively).
534 .. hint::
536 Files include regular files and symbolic links.
538 :returns: Total number of files.
539 """
540 return sum(d.TotalFileCount for d in self._subdirectories.values()) + len(self._files) + len(self._symbolicLinks)
542 @readonly
543 def TotalRegularFileCount(self) -> int:
544 """
545 Read-only property to return the total number of regular files in all child hierarchy levels (recursively).
547 :returns: Total number of regular files.
548 """
549 return sum(d.TotalRegularFileCount for d in self._subdirectories.values()) + len(self._files)
551 @readonly
552 def TotalSymbolicLinkCount(self) -> int:
553 """
554 Read-only property to return the total number of symbolic links in all child hierarchy levels (recursively).
556 :returns: Total number of symbolic links.
557 """
558 return sum(d.TotalSymbolicLinkCount for d in self._subdirectories.values()) + len(self._symbolicLinks)
560 @readonly
561 def TotalSubdirectoryCount(self) -> int:
562 """
563 Read-only property to return the total number of subdirectories in all child hierarchy levels (recursively).
565 :returns: Total number of subdirectories.
566 """
567 return len(self._subdirectories) + sum(d.TotalSubdirectoryCount for d in self._subdirectories.values())
569 @readonly
570 def Subdirectories(self) -> Generator["Directory", None, None]:
571 """
572 Iterate all direct subdirectories of the directory.
574 :returns: A generator to iterate all direct subdirectories.
575 """
576 return (d for d in self._subdirectories.values())
578 @readonly
579 def Files(self) -> Generator["Filename | SymbolicLink", None, None]:
580 """
581 Iterate all direct files of the directory.
583 .. hint::
585 Files include regular files and symbolic links.
587 :returns: A generator to iterate all direct files.
588 """
589 return (f for f in chain(self._files.values(), self._symbolicLinks.values()))
591 @readonly
592 def RegularFiles(self) -> Generator["Filename", None, None]:
593 """
594 Iterate all direct regular files of the directory.
596 :returns: A generator to iterate all direct regular files.
597 """
598 return (f for f in self._files.values())
600 @readonly
601 def SymbolicLinks(self) -> Generator["SymbolicLink", None, None]:
602 """
603 Iterate all direct symbolic links of the directory.
605 :returns: A generator to iterate all direct symbolic links.
606 """
607 return (l for l in self._symbolicLinks.values())
609 @readonly
610 def Path(self) -> Path:
611 """
612 Read-only property to access the equivalent Path instance for accessing the represented directory.
614 :returns: Path to the directory.
615 :raises FilesystemException: If no parent is set.
616 """
617 if self._path is not None:
618 return self._path
620 if self._parent is None:
621 raise FilesystemException(f"No parent or root set for directory.")
623 self._path = self._parent.Path / self._name
624 return self._path
626 @readonly
627 def ScanDuration(self) -> float:
628 """
629 Read-only property to access the time needed to scan a directory structure including all subelements (recursively).
631 :returns: The scan duration in seconds.
632 :raises FilesystemException: If the directory was not scanned.
633 """
634 if self._scanDuration is None:
635 raise FilesystemException(f"Directory was not scanned, yet.")
637 return self._scanDuration
639 @readonly
640 def AggregateDuration(self) -> float:
641 """
642 Read-only property to access the time needed to aggregate the directory's and subelement's properties (recursively).
644 :returns: The aggregation duration in seconds.
645 :raises FilesystemException: If the directory properties were not aggregated.
646 """
647 if self._scanDuration is None:
648 raise FilesystemException(f"Directory properties were not aggregated, yet.")
650 return self._aggregateDuration
652 def __hash__(self) -> int:
653 return hash(id(self))
655 def IterateDirectories(self) -> Generator["Directory", None, None]:
656 # pre-order
657 for directory in self._subdirectories.values():
658 yield directory
659 yield from directory.IterateDirectories()
661 def IterateFiles(self) -> Generator[Element, None, None]:
662 # post-order
663 for directory in self._subdirectories.values():
664 yield from directory.IterateFiles()
666 yield from self._files.values()
667 yield from self._symbolicLinks.values()
669 def Copy(self, parent: Nullable["Directory"] = None) -> "Directory":
670 """
671 Copy the directory structure including all subelements and link it to the given parent.
673 .. hint::
675 Statistics like aggregated directory size are copied too. |br|
676 There is no rescan or repeated aggregation needed.
678 :param parent: The parent element of the copied directory.
679 :returns: A deep copy of the directory structure.
680 """
681 dir = Directory(self._name, parent=parent)
682 dir._size = self._size
684 for subdir in self._subdirectories.values():
685 subdir.Copy(dir)
687 for file in self._files.values():
688 file.Copy(dir)
690 for link in self._symbolicLinks.values():
691 link.Copy(dir)
693 return dir
695 def Collapse(self, func: Callable[["Directory"], bool]) -> bool:
696 # if len(self._subdirectories) == 0 or all(subdir.Collapse(func) for subdir in self._subdirectories.values()):
697 if len(self._subdirectories) == 0:
698 if func(self):
699 # print(f"collapse 1 {self.Path}")
700 self._collapsed = True
701 self._subdirectories.clear()
702 self._files.clear()
703 self._symbolicLinks.clear()
705 return True
706 else:
707 return False
709 # if all(subdir.Collapse(func) for subdir in self._subdirectories.values())
710 collapsible = True
711 for subdir in self._subdirectories.values():
712 result = subdir.Collapse(func)
713 collapsible = collapsible and result
715 if collapsible:
716 # print(f"collapse 2 {self.Path}")
717 self._collapsed = True
718 self._subdirectories.clear()
719 self._files.clear()
720 self._symbolicLinks.clear()
722 return True
723 else:
724 return False
726 def ToTree(self, format: Nullable[Callable[[Node], str]] = None) -> Node:
727 """
728 Convert the directory to a :class:`~pyTooling.Tree.Node`.
730 The node's :attr:`~pyTooling.Tree.Node.Value` field contains a reference to the directory. Additional data is
731 attached to the node's key-value store:
733 ``kind``
734 The node's kind. See :class:`NodeKind`.
735 ``size``
736 The directory's aggregated size.
738 :param format: A user defined formatting function for tree nodes.
739 :returns: A tree node representing this directory.
740 """
741 if format is None:
742 def format(node: Node) -> str:
743 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}"
745 directoryNode = Node(
746 value=self,
747 keyValuePairs={
748 "kind": NodeKind.File,
749 "size": self._size
750 },
751 format=format
752 )
753 directoryNode.AddChildren(
754 e.ToTree(format) for e in chain(self._subdirectories.values()) #, self._files.values(), self._symbolicLinks.values())
755 )
757 return directoryNode
759 def __eq__(self, other) -> bool:
760 """
761 Compare two Directory instances for equality.
763 :param other: Parameter to compare against.
764 :returns: ``True``, if both directories and all its subelements are equal.
765 :raises TypeError: If parameter ``other`` is not of type :class:`Directory`.
766 """
767 if not isinstance(other, Directory):
768 ex = TypeError("Parameter 'other' is not of type Directory.")
769 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.")
770 raise ex
772 if not all(dir1 == dir2 for _, dir1, dir2 in zipdicts(self._subdirectories, other._subdirectories)):
773 return False
775 if not all(file1 == file2 for _, file1, file2 in zipdicts(self._files, other._files)):
776 return False
778 if not all(link1 == link2 for _, link1, link2 in zipdicts(self._symbolicLinks, other._symbolicLinks)):
779 return False
781 return True
783 def __ne__(self, other: Any) -> bool:
784 """
785 Compare two Directory instances for inequality.
787 :param other: Parameter to compare against.
788 :returns: ``True``, if both directories and all its subelements are unequal.
789 :raises TypeError: If parameter ``other`` is not of type :class:`Directory`.
790 """
791 return not self.__eq__(other)
793 def __repr__(self) -> str:
794 return f"Directory: {self.Path}"
796 def __str__(self) -> str:
797 return self._name
800@export
801class Filename(Element[Directory]):
802 """
803 Represents a filename in the filesystem, but not the file storage object (:class:`File`).
805 .. hint::
807 Filename and file storage are represented by two classes, which allows multiple names (hard links) per file storage
808 object.
809 """
810 _file: Nullable["File"]
812 def __init__(
813 self,
814 name: str,
815 file: Nullable["File"] = None,
816 parent: Nullable[Directory] = None
817 ) -> None:
818 """
819 Initialize the filename with name, file (storage) object and parent reference.
821 :param name: Name of the file.
822 :param size: Optional file (storage) object.
823 :param parent: Optional parent reference.
824 """
825 super().__init__(name, None, parent)
827 if file is None: 827 ↛ 830line 827 didn't jump to line 830 because the condition on line 827 was always true
828 self._file = None
829 else:
830 if not isinstance(file, File):
831 ex = TypeError("Parameter 'file' is not of type 'File'.")
832 ex.add_note(f"Got type '{getFullyQualifiedName(file)}'.")
833 raise ex
835 self._file = file
836 file._parents.append(self)
838 if parent is not None:
839 parent._files[name] = self
841 if parent._root is not None:
842 self._root = parent._root
844 @Element.Root.setter
845 def Root(self, value: "Root") -> None:
846 Element.Root.fset(self, value)
848 if self._file is not None: 848 ↛ exitline 848 didn't return from function 'Root' because the condition on line 848 was always true
849 self._file.Root = value
851 @Element.Parent.setter
852 def Parent(self, value: _ParentType) -> None:
853 Element.Parent.fset(self, value)
855 value._files[self._name] = self
857 if isinstance(value, Root): 857 ↛ 858line 857 didn't jump to line 858 because the condition on line 857 was never true
858 self.Root = value
860 @readonly
861 def File(self) -> Nullable["File"]:
862 """
863 Read-only property to access the file this filename is linked to (:attr:`_file`).
865 :returns: The linked file, or ``None`` if the filename isn't linked yet.
866 """
867 return self._file
869 @readonly
870 def Size(self) -> int:
871 """
872 Read-only property to access the size of the linked file.
874 :returns: Size of the linked file in bytes.
875 :raises ToolingException: If the filename isn't linked to a file object.
876 """
877 if self._file is None:
878 raise ToolingException(f"Filename isn't linked to a File object.")
880 return self._file._size
882 @readonly
883 def Path(self) -> Path:
884 """
885 Read-only property to return the filename's absolute path.
887 The path is computed from the parent directory's path and the filename.
889 :returns: Absolute path of the file.
890 :raises ToolingException: If the filename has no parent object.
891 """
892 if self._parent is None:
893 raise ToolingException(f"Filename has no parent object.")
895 return self._parent.Path / self._name
897 def __hash__(self) -> int:
898 return hash(id(self))
900 def Copy(self, parent: Directory) -> "Filename":
901 fileID = self._file._id
903 if fileID in parent._root._ids:
904 file = parent._root._ids[fileID]
905 else:
906 fileSize = self._file._size
907 file = File(fileID, fileSize)
909 parent._root._ids[fileID] = file
911 return Filename(self._name, file, parent=parent)
913 def ToTree(self) -> Node:
914 def format(node: Node) -> str:
915 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}"
917 fileNode = Node(
918 value=self,
919 keyValuePairs={
920 "kind": NodeKind.File,
921 "size": self._size
922 },
923 format=format
924 )
926 return fileNode
928 def __eq__(self, other) -> bool:
929 """
930 Compare two Filename instances for equality.
932 :param other: Parameter to compare against.
933 :returns: ``True``, if both filenames are equal.
934 :raises TypeError: If parameter ``other`` is not of type :class:`Filename`.
935 """
936 if not isinstance(other, Filename):
937 ex = TypeError("Parameter 'other' is not of type 'Filename'.")
938 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.")
939 raise ex
941 return self._name == other._name and self.Size == other.Size
943 def __ne__(self, other: Any) -> bool:
944 """
945 Compare two Filename instances for inequality.
947 :param other: Parameter to compare against.
948 :returns: ``True``, if both filenames are unequal.
949 :raises TypeError: If parameter ``other`` is not of type :class:`Filename`.
950 """
951 if not isinstance(other, Filename):
952 ex = TypeError("Parameter 'other' is not of type 'Filename'.")
953 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.")
954 raise ex
956 return self._name != other._name or self.Size != other.Size
958 def __repr__(self) -> str:
959 return f"File: {self.Path}"
961 def __str__(self) -> str:
962 return self._name
965@export
966class SymbolicLink(Element[Directory]):
967 _target: Path
968 _isConnected: bool
969 _isBroken: Nullable[bool]
970 _isOutOfRange: Nullable[bool]
972 def __init__(
973 self,
974 name: str,
975 target: Path,
976 parent: Nullable[Directory]
977 ) -> None:
978 super().__init__(name, None, parent)
980 if target is None:
981 raise ValueError(f"Parameter 'target' is None.")
982 elif not isinstance(target, Path):
983 ex = TypeError("Parameter 'target' is not of type 'Path'.")
984 ex.add_note(f"Got type '{getFullyQualifiedName(target)}'.")
985 raise ex
987 self._target = target
988 self._isConnected = False
989 self._isBroken = None
990 self._isOutOfRange = None
992 if parent is not None:
993 parent._symbolicLinks[name] = self
995 if parent._root is not None:
996 self._root = parent._root
998 @readonly
999 def Path(self) -> Path:
1000 """
1001 Read-only property to return the symbolic link's path.
1003 The path is computed from the parent directory's path and the link's name.
1005 :returns: Path of the symbolic link.
1006 """
1007 return self._parent.Path / self._name
1009 @readonly
1010 def Target(self) -> Path:
1011 """
1012 Read-only property to access the path this symbolic link points to (:attr:`_target`).
1014 :returns: Target path of the symbolic link.
1015 """
1016 return self._target
1018 @readonly
1019 def IsConnected(self) -> bool:
1020 """
1021 Check if the symbolic link was resolved to an element within the scanned filesystem (:attr:`_isConnected`).
1023 :returns: ``True``, if the link's target was found and connected.
1024 """
1025 return self._isConnected
1027 @readonly
1028 def IsBroken(self) -> Nullable[bool]:
1029 """
1030 Check if the symbolic link points to a non-existing target (:attr:`_isBroken`).
1032 :returns: ``True``, if the target doesn't exist. ``None``, if the link wasn't resolved yet.
1033 """
1034 return self._isBroken
1036 @readonly
1037 def IsOutOfRange(self) -> Nullable[bool]:
1038 """
1039 Check if the symbolic link points outside the scanned filesystem (:attr:`_isOutOfRange`).
1041 :returns: ``True``, if the target lies outside the scanned root. ``None``, if the link wasn't resolved yet.
1042 """
1043 return self._isOutOfRange
1045 def __hash__(self) -> int:
1046 return hash(id(self))
1048 def Copy(self, parent: Directory) -> "SymbolicLink":
1049 return SymbolicLink(self._name, self._target, parent=parent)
1051 def ToTree(self) -> Node:
1052 def format(node: Node) -> str:
1053 return f"{node['size'] * 1e-6:7.1f} MiB {node._value.Name}"
1055 symbolicLinkNode = Node(
1056 value=self,
1057 keyValuePairs={
1058 "kind": NodeKind.SymbolicLink,
1059 "size": self._size
1060 },
1061 format=format
1062 )
1064 return symbolicLinkNode
1066 def __eq__(self, other) -> bool:
1067 """
1068 Compare two SymbolicLink instances for equality.
1070 :param other: Parameter to compare against.
1071 :returns: ``True``, if both symbolic links are equal.
1072 :raises TypeError: If parameter ``other`` is not of type :class:`SymbolicLink`.
1073 """
1074 if not isinstance(other, SymbolicLink):
1075 ex = TypeError("Parameter 'other' is not of type 'SymbolicLink'.")
1076 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.")
1077 raise ex
1079 return self._name == other._name and self._target == other._target
1081 def __ne__(self, other: Any) -> bool:
1082 """
1083 Compare two SymbolicLink instances for inequality.
1085 :param other: Parameter to compare against.
1086 :returns: ``True``, if both symbolic links are unequal.
1087 :raises TypeError: If parameter ``other`` is not of type :class:`SymbolicLink`.
1088 """
1089 if not isinstance(other, SymbolicLink):
1090 ex = TypeError("Parameter 'other' is not of type 'SymbolicLink'.")
1091 ex.add_note(f"Got type '{getFullyQualifiedName(other)}'.")
1092 raise ex
1094 return self._name != other._name or self._target != other._target
1096 def __repr__(self) -> str:
1097 return f"SymLink: {self.Path} -> {self._target}"
1099 def __str__(self) -> str:
1100 return self._name
1103@export
1104class Root(Directory):
1105 """
1106 A **Root** represents the root-directory in the filesystem, which contains subdirectories, regular files and symbolic links.
1107 """
1108 _ids: Dict[int, "File"] #: Dictionary of file identifier - file objects pairs found while scanning the directory structure.
1109 _brokenSymbolicLinks: List[SymbolicLink] #: Broken symbolic links (target doesn't exist).
1110 _unconnectedSymbolicLinks: List[SymbolicLink] #: Symbolic links which couldn't be connected to their target (out of scope).
1112 def __init__(
1113 self,
1114 rootDirectory: Path,
1115 collectSubdirectories: bool = True
1116 ) -> None:
1117 if rootDirectory is None: 1117 ↛ 1118line 1117 didn't jump to line 1118 because the condition on line 1117 was never true
1118 raise ValueError(f"Parameter 'rootDirectory' is None.")
1119 elif not isinstance(rootDirectory, Path): 1119 ↛ 1120line 1119 didn't jump to line 1120 because the condition on line 1119 was never true
1120 raise TypeError(f"Parameter 'rootDirectory' is not of type 'Path'.")
1121 elif not rootDirectory.exists(): 1121 ↛ 1122line 1121 didn't jump to line 1122 because the condition on line 1121 was never true
1122 raise ToolingException(f"Path '{rootDirectory}' doesn't exist.") from FileNotFoundError(rootDirectory)
1124 self._ids = {}
1125 self._brokenSymbolicLinks = []
1126 self._unconnectedSymbolicLinks = []
1128 super().__init__(rootDirectory.name)
1129 self._root = self
1130 self._path = rootDirectory
1132 if collectSubdirectories: 1132 ↛ 1133line 1132 didn't jump to line 1133 because the condition on line 1132 was never true
1133 self.CollectSubdirectories()
1134 self.ResolveSymbolicLinks()
1136 @readonly
1137 def Path(self) -> Path:
1138 """
1139 Read-only property to access the path of the filesystem statistics root.
1141 :returns: Path to the root of the filesystem statistics root directory.
1142 """
1143 return self._path
1145 @readonly
1146 def BrokenSymbolicLinks(self) -> List[SymbolicLink]:
1147 """
1148 Read-only property to access all symbolic links with a non-existing target (:attr:`_brokenSymbolicLinks`).
1150 :returns: List of broken symbolic links.
1151 """
1152 return self._brokenSymbolicLinks
1154 @readonly
1155 def UnconnectedSymbolicLinks(self) -> List[SymbolicLink]:
1156 """
1157 Read-only property to access all symbolic links that couldn't be resolved within the scanned filesystem (:attr:`_unconnectedSymbolicLinks`).
1159 :returns: List of unconnected symbolic links.
1160 """
1161 return self._unconnectedSymbolicLinks
1163 @readonly
1164 def TotalHardLinkCount(self) -> int:
1165 """
1166 Read-only property to return the accumulated number of hardlinks to multiply-linked files.
1168 Every file storage object referenced by more than one directory entry contributes its number of
1169 directory entries.
1171 :returns: Sum of directory entries over all hardlinked files.
1172 """
1173 return sum(l for f in self._ids.values() if (l := len(f._parents)) > 1)
1175 @readonly
1176 def TotalHardLinkCount2(self) -> int:
1177 """
1178 Read-only property to return the number of file storage objects that are hardlinked.
1180 In contrast to :attr:`TotalHardLinkCount`, every hardlinked file contributes ``1``, regardless of how
1181 many directory entries reference it.
1183 :returns: Number of files referenced by more than one directory entry.
1184 """
1185 return sum(1 for f in self._ids.values() if len(f._parents) > 1)
1187 @readonly
1188 def TotalHardLinkCount3(self) -> int:
1189 """
1190 Read-only property to return the number of file storage objects that are **not** hardlinked.
1192 .. attention::
1194 Despite the name, this counts files referenced by exactly *one* directory entry.
1196 :returns: Number of files referenced by exactly one directory entry.
1197 """
1198 return sum(1 for f in self._ids.values() if len(f._parents) == 1)
1200 @readonly
1201 def Size2(self) -> int:
1202 """
1203 Read-only property to return the accumulated size of all hardlinked files, counted once each.
1205 :returns: Sum of sizes over all files referenced by more than one directory entry.
1206 """
1207 return sum(f._size for f in self._ids.values() if len(f._parents) > 1)
1209 @readonly
1210 def Size3(self) -> int:
1211 """
1212 Read-only property to return the accumulated size of all hardlinked files, counted per directory entry.
1214 In contrast to :attr:`Size2`, a file's size is multiplied by the number of directory entries
1215 referencing it, so it reflects the size a filesystem without hardlink support would need.
1217 :returns: Sum of sizes over all hardlinked files, weighted by their number of directory entries.
1218 """
1219 return sum(f._size * len(f._parents) for f in self._ids.values() if len(f._parents) > 1)
1221 @readonly
1222 def TotalUniqueFileCount(self) -> int:
1223 """
1224 Read-only property to return the number of distinct file storage objects, counting hardlinks to the same content once.
1226 :returns: Number of unique files.
1227 """
1228 return len(self._ids)
1230 def RegisterBrokenSymbolicLink(self, symLink: SymbolicLink) -> None:
1231 symLink._isBroken = True
1232 self._brokenSymbolicLinks.append(symLink)
1234 def RegisterUnconnectedSymbolicLink(self, symLink: SymbolicLink) -> None:
1235 symLink._isOutOfRange = True
1236 self._unconnectedSymbolicLinks.append(symLink)
1238 def Copy(self) -> "Root":
1239 """
1240 Copy the directory structure including all subelements and link it to the given parent.
1242 The duration for the deep copy process is provided in :attr:`ScanDuration`
1244 .. hint::
1246 Statistics like aggregated directory size are copied too. |br|
1247 There is no rescan or repeated aggregation needed.
1249 :returns: A deep copy of the directory structure.
1250 """
1251 with Stopwatch() as sw:
1252 root = Root(self._path, False)
1253 root._size = self._size
1255 for subdir in self._subdirectories.values():
1256 subdir.Copy(root)
1258 for file in self._files.values():
1259 file.Copy(root)
1261 for link in self._symbolicLinks.values():
1262 link.Copy(root)
1264 root._scanDuration = sw.Duration
1265 root._aggregateDuration = 0.0
1267 return root
1269 def __repr__(self) -> str:
1270 return f"Root: {self.Path} (dirs: {self.TotalSubdirectoryCount}, files: {self.TotalRegularFileCount}, symlinks: {self.TotalSymbolicLinkCount})"
1272 def __str__(self) -> str:
1273 return self._name
1276@export
1277class File(Base):
1278 """
1279 A **File** represents a file storage object in the filesystem, which is accessible by one or more :class:`Filename` objects.
1281 Each file has an internal id, which is associated to a unique ID within the host's filesystem.
1282 """
1283 _id: int #: Unique (host internal) file object ID)
1284 _parents: List[Filename] #: List of reverse references to :class:`Filename` objects.
1286 def __init__(
1287 self,
1288 id: int,
1289 size: int,
1290 parent: Nullable[Filename] = None
1291 ) -> None:
1292 """
1293 Initialize the File storage object with an ID, size and parent reference.
1295 :param id: Unique ID of the file object.
1296 :param size: Size of the file object.
1297 :param parent: Optional parent reference.
1298 """
1299 if id is None: 1299 ↛ 1300line 1299 didn't jump to line 1300 because the condition on line 1299 was never true
1300 raise ValueError(f"Parameter 'id' is None.")
1301 elif not isinstance(id, int): 1301 ↛ 1302line 1301 didn't jump to line 1302 because the condition on line 1301 was never true
1302 ex = TypeError("Parameter 'id' is not of type 'int'.")
1303 ex.add_note(f"Got type '{getFullyQualifiedName(id)}'.")
1304 raise ex
1306 self._id = id
1308 if parent is None:
1309 super().__init__(size, None)
1310 self._parents = []
1311 elif isinstance(parent, Filename): 1311 ↛ 1316line 1311 didn't jump to line 1316 because the condition on line 1311 was always true
1312 super().__init__(size, parent._root)
1313 self._parents = [parent]
1314 parent._file = self
1315 else:
1316 ex = TypeError("Parameter 'parent' is not of type 'Filename'.")
1317 ex.add_note(f"Got type '{getFullyQualifiedName(parent)}'.")
1318 raise ex
1320 @readonly
1321 def ID(self) -> int:
1322 """
1323 Read-only property to access the file object's unique identifier.
1325 :returns: Unique file object identifier.
1326 """
1327 return self._id
1329 @readonly
1330 def Parents(self) -> List[Filename]:
1331 """
1332 Read-only property to access the list of filenames using the same file storage object.
1334 .. hint::
1336 This allows to check if a file object has multiple filenames a.k.a hardlinks.
1338 :returns: List of filenames for the file storage object.
1339 """
1340 return self._parents
1342 def AddParent(self, filename: Filename) -> None:
1343 """
1344 Add another parent reference to a :class:`Filename`.
1346 :param filename: Reference to a filename object.
1347 """
1348 if filename is None: 1348 ↛ 1349line 1348 didn't jump to line 1349 because the condition on line 1348 was never true
1349 raise ValueError(f"Parameter 'filename' is None.")
1350 elif not isinstance(filename, Filename): 1350 ↛ 1351line 1350 didn't jump to line 1351 because the condition on line 1350 was never true
1351 ex = TypeError("Parameter 'filename' is not of type 'Filename'.")
1352 ex.add_note(f"Got type '{getFullyQualifiedName(filename)}'.")
1353 raise ex
1354 elif filename._file is not None: 1354 ↛ 1355line 1354 didn't jump to line 1355 because the condition on line 1354 was never true
1355 raise ToolingException(f"Filename is already referencing an other file object ({filename._file._id}).")
1357 self._parents.append(filename)
1358 filename._file = self
1360 if filename._root is not None:
1361 self._root = filename._root