我们从Python开源项目中，提取了以下14个代码示例，用于说明如何使用pyinotify.ProcessEvent()。
def setup_inotify(configuration): class EventHandler(pyinotify.ProcessEvent): def process_default(self, event): basename = os.path.basename(event.pathname) if basename == 'node.json': log("node.json changed") configuration.parse_node_json() elif basename == 'config.json': log("config.json changed!") configuration.parse_config_json() elif basename.endswith('.py'): abort_service("python file changed") wm = pyinotify.WatchManager() notifier = pyinotify.ThreadedNotifier(wm, EventHandler()) notifier.daemon = True notifier.start() wm.add_watch('.', pyinotify.IN_MOVED_TO)
def watch_config_changes(config_path, queues, nuimo_apps, processes, ha_api_url, ble_adapter_name): class ModificationHandler(pyinotify.ProcessEvent): def process_IN_CLOSE_WRITE(self, event): if hasattr(event, 'pathname') and event.pathname == config_path: logger.info("Config file was changed, reloading it...") update_from_config_file(config_path, queues, nuimo_apps, processes, ha_api_url, ble_adapter_name) handler = ModificationHandler() watch_manager = pyinotify.WatchManager() notifier = pyinotify.Notifier(watch_manager, handler) # IN_CLOSE_WRITE is fired when the file was closed after modification # in opposite to IN_MODIFY which is called for each partial write watch_manager.add_watch(config_path, pyinotify.IN_CLOSE_WRITE) logger.info("Listening to changes of: %s", config_path) notifier.loop() logger.info("Stopped listening to changes of: %s", config_path)
def wait_pyinotify(self, bld): class PE(w_pyinotify.ProcessEvent): def stop(self, event): self.notif.ev = True self.notif.stop() raise ValueError("stop for delete") process_IN_DELETE = stop process_IN_CLOSE = stop process_default = stop proc = PE() wm = w_pyinotify.WatchManager() notif = w_pyinotify.Notifier(wm, proc) proc.notif = notif # well, we should add all the folders to watch here for x in self.enumerate(bld.srcnode): wm.add_watch(x, w_pyinotify.IN_DELETE | w_pyinotify.IN_CLOSE_WRITE) try: # pyinotify uses an infinite loop ... not too nice, so we have to use an exception notif.loop() except ValueError: pass if not hasattr(notif, 'ev'): raise KeyboardInterrupt
def decrypt_to_fifo(_input, _output, _mode, _force, text=None): def write_to_fifo(notifier): st = os.stat(_output) if not stat.S_ISFIFO(st.st_mode): raise click.ClickException('%s is not a FIFO!' % _output) if (st.st_mode & 0o777 != _mode): raise click.ClickException('mode has changed on %s!' % _output) if _input != '-': f_in = open(_input, 'rb') text = f_in.read() f_in.close() try: f = open(_output, 'w') text = decrypt_string(text) f.write(text) f.close() except IOError: pass finally: text = str(0x00) * len(text) wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, pyinotify.ProcessEvent) wm.add_watch(_output, pyinotify.IN_CLOSE_NOWRITE) notifier.loop(callback=write_to_fifo)
def inotify_code_changed(): """ Checks for changed code using inotify. After being called it blocks until a change event has been fired. """ class EventHandler(pyinotify.ProcessEvent): modified_code = None def process_default(self, event): if event.path.endswith('.mo'): EventHandler.modified_code = I18N_MODIFIED else: EventHandler.modified_code = FILE_MODIFIED wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, EventHandler()) def update_watch(sender=None, **kwargs): if sender and getattr(sender, 'handles_files', False): # No need to update watches when request serves files. # (sender is supposed to be a django.core.handlers.BaseHandler subclass) return mask = ( pyinotify.IN_MODIFY | pyinotify.IN_DELETE | pyinotify.IN_ATTRIB | pyinotify.IN_MOVED_FROM | pyinotify.IN_MOVED_TO | pyinotify.IN_CREATE | pyinotify.IN_DELETE_SELF | pyinotify.IN_MOVE_SELF ) for path in gen_filenames(only_new=True): wm.add_watch(path, mask) # New modules may get imported when a request is processed. request_finished.connect(update_watch) # Block until an event happens. update_watch() notifier.check_events(timeout=None) notifier.read_events() notifier.process_events() notifier.stop() # If we are here the code must have changed. return EventHandler.modified_code
def inotify_code_changed(): """ Checks for changed code using inotify. After being called it blocks until a change event has been fired. """ class EventHandler(pyinotify.ProcessEvent): modified_code = None def process_default(self, event): if event.path.endswith('.mo'): EventHandler.modified_code = I18N_MODIFIED else: EventHandler.modified_code = FILE_MODIFIED wm = pyinotify.WatchManager() notifier = pyinotify.Notifier(wm, EventHandler()) def update_watch(sender=None, **kwargs): if sender and getattr(sender, 'handles_files', False): # No need to update watches when request serves files. # (sender is supposed to be a django.core.handlers.BaseHandler subclass) return mask = ( pyinotify.IN_MODIFY | pyinotify.IN_DELETE | pyinotify.IN_ATTRIB | pyinotify.IN_MOVED_FROM | pyinotify.IN_MOVED_TO | pyinotify.IN_CREATE ) for path in gen_filenames(only_new=True): wm.add_watch(path, mask) # New modules may get imported when a request is processed. request_finished.connect(update_watch) # Block until an event happens. update_watch() notifier.check_events(timeout=None) notifier.read_events() notifier.process_events() notifier.stop() # If we are here the code must have changed. return EventHandler.modified_code
