lib/bup/io.py

   1
   2 import mmap as py_mmap
   3
   4 from bup.compat import pending_raise
   5
   6
   7 def byte_stream(file):
   8     return file.buffer
   9
  10 def path_msg(x):
  11     """Return a string representation of a path."""
  12     # FIXME: configurability (might git-config quotePath be involved?)
  13     return x.decode(errors='backslashreplace')
  14
  15
  16 assert not hasattr(py_mmap.mmap, '__del__')
  17 if hasattr(py_mmap.mmap, '__enter__'):
  18     assert hasattr(py_mmap.mmap, '__exit__')
  19
  20 class mmap(py_mmap.mmap):
  21     '''mmap.mmap wrapper that detects and complains about any instances
  22     that aren't explicitly closed.
  23
  24     '''
  25
  26     def __init__(self, *args, **kwargs):
  27         self._bup_closed = True
  28         # Silence deprecation warnings.  mmap's current parent is
  29         # object, which accepts no params and as of at least 2.7
  30         # warns about them.
  31         if py_mmap.mmap.__init__ is not object.__init__:
  32             super(mmap, self).__init__(self, *args, **kwargs)
  33         self._bup_closed = False
  34
  35     def close(self):
  36         self._bup_closed = True
  37         super(mmap, self).close()
  38
  39     if hasattr(py_mmap.mmap, '__enter__'):
  40         def __enter__(self):
  41             super(mmap, self).__enter__()
  42             return self
  43         def __exit__(self, type, value, traceback):
  44             # Don't call self.close() when the parent has its own __exit__;
  45             # defer to it.
  46             self._bup_closed = True
  47             result = super(mmap, self).__exit__(type, value, traceback)
  48             return result
  49     else:
  50         def __enter__(self):
  51             return self
  52         def __exit__(self, type, value, traceback):
  53             with pending_raise(value, rethrow=False):
  54                 self.close()
  55
  56     def __del__(self):
  57         assert self._bup_closed