¡@
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, remove=os.unlink): """Delete a file, but ignore file not found error. :param path: File to delete :param remove: Optional function to remove passed path """ try: remove(path) except OSError as e: if e.errno != errno.ENOENT: raise @contextlib.contextmanager
**** CubicPower OpenStack Study **** def remove_path_on_error(path, remove=delete_if_exists): """Protect code that wants to operate on PATH atomically. Any exception will cause PATH to be removed. :param path: File to work with :param remove: Optional function to remove passed path """ try: yield except Exception: with excutils.save_and_reraise_exception(): remove(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)
**** CubicPower OpenStack Study **** def write_to_tempfile(content, path=None, suffix='', prefix='tmp'): """Create temporary file or use existing file. This util is needed for creating temporary file with specified content, suffix and prefix. If path is not None, it will be used for writing content. If the path doesn't exist it'll be created. :param content: content for temporary file. :param path: same as parameter 'dir' for mkstemp :param suffix: same as parameter 'suffix' for mkstemp :param prefix: same as parameter 'prefix' for mkstemp For example: it can be used in database tests for creating configuration files. """ if path: ensure_tree(path) (fd, path) = tempfile.mkstemp(suffix=suffix, dir=path, prefix=prefix) try: os.write(fd, content) finally: os.close(fd) return path
|