我们从Python开源项目中，提取了以下12个代码示例，用于说明如何使用pyinotify.IN_MOVED_TO。
def __init__(self, rt): ManagerPlugin.__init__(self, rt) self.git_repo = GitRepo(directory=self.rt.paths.skills, url='https://github.com/MatthewScholefield/mycroft-light.git', branch='skills', update_freq=1) self.blacklist = self.config['blacklist'] sys.path.append(self.rt.paths.skills) GroupPlugin.__init__(self, SkillPlugin, 'mycroft.skills', '_skill') for cls in self._classes.values(): cls.rt = rt self.init_plugins() # The watch manager stores the watches and provides operations on watches wm = pyinotify.WatchManager() mask = pyinotify.IN_MODIFY | pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MOVED_TO skills_dir = self.rt.paths.skills handler = EventHandler(self) notifier = pyinotify.ThreadedNotifier(wm, handler) notifier.daemon = True wdd = wm.add_watch(skills_dir, mask, rec=True) notifier.start()
def __init__(self, listen_host, listen_port, server_key, server_cert, ca_cert, image_path, image_types, exec_handlers, static_paths): websockets = WebSockets() wm = pyinotify.WatchManager() inotify_handler = INotifyHandler(websockets) self._notifier = pyinotify.Notifier(wm, inotify_handler) for image_type in image_types: type_path = os.path.join(image_path, image_type) wm.add_watch(type_path, pyinotify.IN_MOVED_TO) exec_handlers = dict(x.split('=', 1) for x in (exec_handlers or [])) static_paths = dict(x.split('=', 1) for x in (static_paths or [])) http_handler = HTTPRequestHandler(image_path, image_types, exec_handlers, static_paths, websockets) self._httpd = geventserver.WSGIServer( (listen_host, listen_port), http_handler, keyfile=server_key, certfile=server_cert, ca_certs=ca_cert, cert_reqs=ssl.CERT_REQUIRED, ssl_version=ssl.PROTOCOL_TLSv1_2)
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_shakemaps_push( working_dir, timeout=None, handler=None, daemon=False): wm = pyinotify.WatchManager() if daemon: notifier = pyinotify.ThreadedNotifier(wm, handler, timeout=timeout) else: notifier = pyinotify.Notifier(wm, handler, timeout=timeout) flags = pyinotify.IN_CREATE | pyinotify.IN_MODIFY | pyinotify.IN_MOVED_TO wm.add_watch(working_dir, flags, rec=True, auto_add=True) return notifier
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
