How to use the watchdog.events.FileModifiedEvent function in watchdog

To help you get started, we’ve selected a few watchdog examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github gorakhargosh / watchdog / tests / test_observer.py View on Github external
def test_unschedule_self(observer):
    """
    Tests that unscheduling a watch from within an event handler correctly
    correctly unregisters emitter and handler without deadlocking.
    """
    class EventHandler(FileSystemEventHandler):
        def on_modified(self, event):
            observer.unschedule(watch)
            unschedule_finished.set()

    unschedule_finished = Event()
    watch = observer.schedule(EventHandler(), '')
    observer.start()

    (emitter,) = observer.emitters
    emitter.queue_event(FileModifiedEvent(''))

    assert unschedule_finished.wait()
    assert len(observer.emitters) == 0
github brickgao / specchio / tests / test_handlers.py View on Github external
def test_on_modified(self, _rsync, _remote_create_folder, _os):
        with mock.patch.object(
                self.handler,
                "update_gitignore"
        ) as _update_gitignore:
            _rsync.return_value = True
            _remote_create_folder.return_value = True
            _os.path.abspath.return_value = "/a/.gitignore"
            _os.path.join.return_value = "/b/a/.gitignore"
            _update_gitignore.return_value = True
            _event = FileModifiedEvent(src_path="/a/.gitignore")
            self.handler.on_modified(_event)
            _rsync.assert_called_once_with(
                dst_ssh=self.handler.dst_ssh,
                src_path="/a/.gitignore",
                dst_path="/b/a/.gitignore"
            )
            _remote_create_folder.assert_called_once_with(
                dst_ssh=self.handler.dst_ssh,
                dst_path="/b/a/"
            )
            _update_gitignore.assert_called_once_with(
                "/a/.gitignore"
            )
github gorakhargosh / watchdog / tests / legacy / test_watchdog_events.py View on Github external
dir_del_event_ignored = DirDeletedEvent('/path/foobar.pyc')
        file_del_event_match = FileDeletedEvent('/path/blah.txt')
        file_del_event_not_match = FileDeletedEvent('/path/foobar')
        file_del_event_ignored = FileDeletedEvent('/path/blah.pyc')

        dir_cre_event_match = DirCreatedEvent('/path/blah.py')
        dir_cre_event_not_match = DirCreatedEvent('/path/foobar')
        dir_cre_event_ignored = DirCreatedEvent('/path/foobar.pyc')
        file_cre_event_match = FileCreatedEvent('/path/blah.txt')
        file_cre_event_not_match = FileCreatedEvent('/path/foobar')
        file_cre_event_ignored = FileCreatedEvent('/path/blah.pyc')

        dir_mod_event_match = DirModifiedEvent('/path/blah.py')
        dir_mod_event_not_match = DirModifiedEvent('/path/foobar')
        dir_mod_event_ignored = DirModifiedEvent('/path/foobar.pyc')
        file_mod_event_match = FileModifiedEvent('/path/blah.txt')
        file_mod_event_not_match = FileModifiedEvent('/path/foobar')
        file_mod_event_ignored = FileModifiedEvent('/path/blah.pyc')

        dir_mov_event_match = DirMovedEvent('/path/blah.py', '/path/blah')
        dir_mov_event_not_match = DirMovedEvent('/path/foobar', '/path/blah')
        dir_mov_event_ignored = DirMovedEvent('/path/foobar.pyc', '/path/blah')
        file_mov_event_match = FileMovedEvent('/path/blah.txt', '/path/blah')
        file_mov_event_not_match = FileMovedEvent('/path/foobar', '/path/blah')
        file_mov_event_ignored = FileMovedEvent('/path/blah.pyc', '/path/blah')

        all_dir_events = [
            dir_mod_event_match,
            dir_mod_event_not_match,
            dir_mod_event_ignored,
            dir_del_event_match,
            dir_del_event_not_match,
github brickgao / specchio / specchio / handlers.py View on Github external
def on_modified(self, event):
        abs_src_path = os.path.abspath(event.src_path)
        isdir = isinstance(event, DirModifiedEvent)
        if self.is_ignore(abs_src_path, isdir):
            return
        if isinstance(event, FileModifiedEvent):
            relative_path = self.get_relative_src_path(event.src_path)
            dst_path = os.path.join(self.dst_path, relative_path)
            # Create folder of file
            dst_folder_path = dst_path[:-len(dst_path.split("/")[-1])]
            # If the file is `.gitignore`, update gitignore dict and list
            if dst_path.split("/")[-1] == ".gitignore":
                logger.info("Update ignore pattern, because changed "
                            "file({}) named `.gitignore` locally".format(
                                abs_src_path
                            ))
                self.update_gitignore(abs_src_path)
            remote_create_folder(dst_ssh=self.dst_ssh,
                                 dst_path=dst_folder_path)
            logger.info("Rsync {} remotely".format(dst_path))
            rsync(dst_ssh=self.dst_ssh, src_path=abs_src_path,
                  dst_path=dst_path)
github CenterForOpenScience / osf-sync / osfoffline / filesystem_manager / sync_local_filesystem_and_db.py View on Github external
raise TypeError
        if db and (not (isinstance(db, Node) or isinstance(db, File))):
            raise TypeError

        event = None

        if local:
            # TODO: find better way of ignoring files we want to ignore
            if local.name.startswith('.') or local.name.endswith('~'):
                return event

        if local and db:
            if self._get_proper_path(local) != self._get_proper_path(db):
                raise IncorrectLocalDBMatch
            if isinstance(db, File) and db.is_file and self._make_hash(local) != db.hash:
                event = FileModifiedEvent(self._get_proper_path(local).full_path)  # create changed event
                # folder modified event cannot happen. It will be a create and delete event.
        elif local is None:
            db_path = self._get_proper_path(db).full_path
            if isinstance(db, File) and db.is_file:
                event = FileDeletedEvent(db_path)  # delete event for file
            else:
                event = DirDeletedEvent(db_path)  # delete event for folder
        elif db is None:
            local_path = self._get_proper_path(local)
            if local_path.is_dir:
                event = DirCreatedEvent(local_path.full_path)
            else:
                event = FileCreatedEvent(local_path.full_path)
        return event
github DocMarty84 / koozic / odoo / service / server.py View on Github external
def dispatch(self, event):
        if isinstance(event, (FileCreatedEvent, FileModifiedEvent, FileMovedEvent)):
            if not event.is_directory:
                path = getattr(event, 'dest_path', event.src_path)
                self.handle_file(path)
github gorakhargosh / watchdog / watchdog / observers / fsevents_observer.py View on Github external
def _dispatch_events_for_path(self, handler, recursive, path):
            with self._lock:
                diff = self._get_directory_snapshot_diff(path, recursive)
                if diff:
                    for path in diff.files_deleted:
                        handler.dispatch(FileDeletedEvent(path, handler))
                    for path in diff.files_modified:
                        handler.dispatch(FileModifiedEvent(path, handler))
                    for path in diff.files_created:
                        handler.dispatch(FileCreatedEvent(path, handler))
                    for path, dest_path in diff.files_moved.items():
                        handler.dispatch(FileMovedEvent(path, dest_path, handler))

                    for path in diff.dirs_modified:
                        handler.dispatch(DirModifiedEvent(path, handler))
                    for path in diff.dirs_deleted:
                        handler.dispatch(DirDeletedEvent(path, handler))
                    for path in diff.dirs_created:
                        handler.dispatch(DirCreatedEvent(path, handler))
                    for path, dest_path in diff.dirs_moved.items():
                        handler.dispatch(DirMovedEvent(path, dest_path, handler))
github gorakhargosh / watchdog / src / watchdog / observers / kqueue.py View on Github external
else:
                    self.queue_event(FileDeletedEvent(src_path))
            elif is_attrib_modified(kev):
                if descriptor.is_directory:
                    self.queue_event(DirModifiedEvent(src_path))
                else:
                    self.queue_event(FileModifiedEvent(src_path))
            elif is_modified(kev):
                if descriptor.is_directory:
                    # When a directory is modified, it may be due to
                    # sub-file/directory renames or new file/directory
                    # creation. We determine all this by comparing
                    # snapshots later.
                    dirs_modified.add(src_path)
                else:
                    self.queue_event(FileModifiedEvent(src_path))
            elif is_renamed(kev):
                # Kqueue does not specify the destination names for renames
                # to, so we have to process these after taking a snapshot
                # of the directory.
                if descriptor.is_directory:
                    dirs_renamed.add(src_path)
                else:
                    files_renamed.add(src_path)
        return files_renamed, dirs_renamed, dirs_modified
github gorakhargosh / watchdog / watchdog / observers / kqueue_observer.py View on Github external
# Modified
                    if watch.is_directory:
                        event = DirModifiedEvent(src_path,
                                                 handlers=self.handlers)
                    else:
                        event = FileModifiedEvent(src_path,
                                                  handlers=self.handlers)
                    self.event_queue.put(event)

                elif is_modified(kev):
                    # Modified
                    if watch.is_directory:
                        dirs_modified.add(src_path)
                    else:
                        self.event_queue.put(
                            FileModifiedEvent(src_path,
                                              handlers=self.handlers))

                elif is_renamed(kev):
                    # Renamed/moved
                    if watch.is_directory:
                        dirs_renamed.add(src_path)
                    else:
                        files_renamed.add(src_path)

            return files_renamed, dirs_renamed, dirs_modified
github spacemanspiff2007 / HABApp / HABApp / runtime / folder_watcher / fileeventhandler.py View on Github external
def on_modified(self, event):
        if not isinstance(event, FileModifiedEvent):
            return None

        if not event.src_path.endswith(self.__file_ending):
            return None

        self._get_func(self.target.reload_file)(Path(event.src_path))