X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=src%2Fpyutils%2Ffiles%2Fdirectory_filter.py;h=243056e9d58191150271536545b0eea5bce36198;hb=HEAD;hp=d7499a2622960449673f283c9426a20d7d3bb251;hpb=993b0992473c12294ed659e52b532e1c8cf9cd1e;p=pyutils.git diff --git a/src/pyutils/files/directory_filter.py b/src/pyutils/files/directory_filter.py index d7499a2..243056e 100644 --- a/src/pyutils/files/directory_filter.py +++ b/src/pyutils/files/directory_filter.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -# © Copyright 2021-2022, Scott Gasch +# © Copyright 2021-2023, Scott Gasch """This module contains two classes meant to help reduce unnecessary disk I/O operations: @@ -30,6 +30,9 @@ class DirectoryFileFilter(object): content to-be-written is identical to the contents of the file on disk allowing calling code to safely skip the write. + Raises: + ValueError: directory doesn't exist + >>> testfile = '/tmp/directory_filter_text_f39e5b58-c260-40da-9448-ad1c3b2a69c2.txt' >>> contents = b'This is a test' >>> with open(testfile, 'wb') as wf: @@ -70,7 +73,7 @@ class DirectoryFileFilter(object): for direntry in os.scandir(self.directory): if direntry.is_file(follow_symlinks=True): mtime = direntry.stat(follow_symlinks=True).st_mtime - path = f'{self.directory}/{direntry.name}' + path = f"{self.directory}/{direntry.name}" self._update_file(path, mtime) def _update_file(self, filename: str, mtime: Optional[float] = None): @@ -87,7 +90,7 @@ class DirectoryFileFilter(object): if self.mtime_by_filename.get(filename, 0) != mtime: md5 = file_utils.get_file_md5(filename) logger.debug( - 'Computed/stored %s\'s MD5 at ts=%.2f (%s)', filename, mtime, md5 + "Computed/stored %s's MD5 at ts=%.2f (%s)", filename, mtime, md5 ) self.mtime_by_filename[filename] = mtime self.md5_by_filename[filename] = md5 @@ -110,11 +113,11 @@ class DirectoryFileFilter(object): """ self._update_file(filename) file_md5 = self.md5_by_filename.get(filename, 0) - logger.debug('%s\'s checksum is %s', filename, file_md5) + logger.debug("%s's checksum is %s", filename, file_md5) mem_hash = hashlib.md5() mem_hash.update(proposed_contents) md5 = mem_hash.hexdigest() - logger.debug('Item\'s checksum is %s', md5) + logger.debug("Item's checksum is %s", md5) return md5 != file_md5 @@ -173,7 +176,9 @@ class DirectoryAllFilesFilter(DirectoryFileFilter): self.md5_by_filename[filename] = md5 self.all_md5s.add(md5) - def apply(self, proposed_contents: Any, ignored_filename: str = None) -> bool: + def apply( + self, proposed_contents: Any, ignored_filename: Optional[str] = None + ) -> bool: """Call this before writing a new file to directory with the proposed_contents to be written and it will return a value that indicates whether the identical contents is already sitting in @@ -196,7 +201,7 @@ class DirectoryAllFilesFilter(DirectoryFileFilter): return md5 not in self.all_md5s -if __name__ == '__main__': +if __name__ == "__main__": import doctest doctest.testmod()