Source code for fs.tempfs

"""Manage filesystems in temporary locations.

A temporary filesytem is stored in a location defined by your OS
(``/tmp`` on linux). The contents are deleted when the filesystem
is closed.

A `TempFS` is a good way of preparing a directory structure in advance,
that you can later copy. It can also be used as a temporary data store.


from __future__ import print_function
from __future__ import unicode_literals

import shutil
import tempfile
import typing

import six

from . import errors
from .osfs import OSFS

if False:  # typing.TYPE_CHECKING
    from typing import Optional, Text

[docs]@six.python_2_unicode_compatible class TempFS(OSFS): """A temporary filesystem on the OS. Arguments: identifier (str): A string to distinguish the directory within the OS temp location, used as part of the directory name. temp_dir (str, optional): An OS path to your temp directory (leave as `None` to auto-detect) auto_clean (bool): If `True` (the default), the directory contents will be wiped on close. ignore_clean_errors (bool): If `True` (the default), any errors in the clean process will be suppressed. If `False`, they will be raised. """ def __init__( self, identifier="__tempfs__", # type: Text temp_dir=None, # type: Optional[Text] auto_clean=True, # type: bool ignore_clean_errors=True, # type: bool ): # type: (...) -> None self.identifier = identifier self._auto_clean = auto_clean self._ignore_clean_errors = ignore_clean_errors self._cleaned = False self.identifier = identifier.replace("/", "-") self._temp_dir = tempfile.mkdtemp(identifier or "fsTempFS", dir=temp_dir) super(TempFS, self).__init__(self._temp_dir) def __repr__(self): # type: () -> Text return "TempFS()" def __str__(self): # type: () -> Text return "<tempfs '{}'>".format(self._temp_dir) def close(self): # type: () -> None if self._auto_clean: self.clean() super(TempFS, self).close()
[docs] def clean(self): # type: () -> None """Clean (delete) temporary files created by this filesystem. """ if self._cleaned: return try: shutil.rmtree(self._temp_dir) except Exception as error: if not self._ignore_clean_errors: raise errors.OperationFailed( msg="failed to remove temporary directory", exc=error ) self._cleaned = True