¡@
OpenStack Study: fileutils.py**** CubicPower OpenStack Study **** def ensure_tree(path): """Create a directory (and any ancestor directories required) :param path: Directory to create """ try: os.makedirs(path) except OSError as exc: if exc.errno == errno.EEXIST: if not os.path.isdir(path): raise else: raise
**** CubicPower OpenStack Study **** def read_cached_file(filename, force_reload=False): """Read from a file if it has been modified. :param force_reload: Whether to reload the file. :returns: A tuple with a boolean specifying if the data is fresh or not. """ global _FILE_CACHE if force_reload and filename in _FILE_CACHE: del _FILE_CACHE[filename] reloaded = False mtime = os.path.getmtime(filename) cache_info = _FILE_CACHE.setdefault(filename, {}) if not cache_info or mtime > cache_info.get('mtime', 0): LOG.debug(_("Reloading cached file %s") % filename) with open(filename) as fap: cache_info['data'] = fap.read() cache_info['mtime'] = mtime reloaded = True return (reloaded, cache_info['data'])
**** CubicPower OpenStack Study **** def delete_if_exists(path): """Delete a file, but ignore file not found error. :param path: File to delete """ try: os.unlink(path) except OSError as e: if e.errno == errno.ENOENT: return else: raise @contextlib.contextmanager
**** CubicPower OpenStack Study **** def remove_path_on_error(path): """Protect code that wants to operate on PATH atomically. Any exception will cause PATH to be removed. :param path: File to work with """ try: yield except Exception: with excutils.save_and_reraise_exception(): delete_if_exists(path)
**** CubicPower OpenStack Study **** def file_open(*args, **kwargs): """Open file see built-in file() documentation for more details Note: The reason this is kept in a separate module is to easily be able to provide a stub module that doesn't alter system state at all (for unit tests) """ return file(*args, **kwargs)
|