Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _walk(self, dir_path, changes, new_files):
for entry in os.scandir(dir_path):
if entry.is_dir():
if self.should_watch_dir(entry):
self._walk(entry.path, changes, new_files)
elif self.should_watch_file(entry):
mtime = entry.stat().st_mtime
new_files[entry.path] = mtime
old_mtime = self.files.get(entry.path)
if not old_mtime:
changes.add((Change.added, entry.path))
elif old_mtime != mtime:
changes.add((Change.modified, entry.path))
def check(self):
changes = set()
new_files = {}
try:
self._walk(str(self.root_path), changes, new_files)
except OSError as e:
# happens when a directory has been deleted between checks
logger.warning('error walking file system: %s %s', e.__class__.__name__, e)
# look for deleted
deleted = self.files.keys() - new_files.keys()
if deleted:
changes |= {(Change.deleted, entry) for entry in deleted}
self.files = new_files
return changes
def _walk(self, dir_path, changes, new_files):
for entry in os.scandir(dir_path):
if entry.is_dir():
if self.should_watch_dir(entry):
self._walk(entry.path, changes, new_files)
elif self.should_watch_file(entry):
mtime = entry.stat().st_mtime
new_files[entry.path] = mtime
old_mtime = self.files.get(entry.path)
if not old_mtime:
changes.add((Change.added, entry.path))
elif old_mtime != mtime:
changes.add((Change.modified, entry.path))