我们从Python开源项目中,提取了以下9个代码示例,用于说明如何使用pyinotify.IN_DELETE_SELF。
def FSMonitor(self): wm = WatchManager() mask = IN_DELETE_SELF | IN_ACCESS notifier = Notifier(wm,EventHandle()) while True: try: notifier.process_events() wm.add_watch(usericonPath,mask,rec=True) if notifier.check_events(): notifier.read_events() except KeyboardInterrupt: notifier.stop() break
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 _reloader_inotify(extra_files=None, interval=None): # Mutated by inotify loop when changes occur. changed = [False] # Setup inotify watches from pyinotify import WatchManager, Notifier # this API changed at one point, support both try: from pyinotify import EventsCodes as ec ec.IN_ATTRIB except (ImportError, AttributeError): import pyinotify as ec wm = WatchManager() mask = ec.IN_DELETE_SELF | ec.IN_MOVE_SELF | ec.IN_MODIFY | ec.IN_ATTRIB def signal_changed(event): if changed[0]: return _log('info', ' * Detected change in %r, reloading' % event.path) changed[:] = [True] for fname in extra_files or (): wm.add_watch(fname, mask, signal_changed) # ... And now we wait... notif = Notifier(wm) try: while not changed[0]: # always reiterate through sys.modules, adding them for fname in _iter_module_files(): wm.add_watch(fname, mask, signal_changed) notif.process_events() if notif.check_events(timeout=interval): notif.read_events() # TODO Set timeout to something small and check parent liveliness finally: notif.stop() sys.exit(3) # currently we always use the stat loop reloader for the simple reason # that the inotify one does not respond to added files properly. Also # it's quite buggy and the API is a mess.