On Mar 15, 2009, at 3:25 PM, Greg Ewing wrote:
with renaming_file("blarg.txt", "w") as f:
...
By chance during the weekend I actually wrote something like that:
from __future__ import with_statement
import os
import codecs
import shutil
import tempfile
from contextlib import contextmanager
TDIR = tempfile.mktemp(dir='/tmp/')
@contextmanager
def topen(filepath, mode='wb', bufsize=-1, encoding=None,
inplace=False, tmpd=TDIR, sync=False):
"""
C{topen} is a transactional version of the Python builtin C{open}
function for file IO. It manages transactionality by using a
temporary file and then moving it to the final position once its
content has been written to disk.
If the mode used to open the file doesn't modify the file, this
function is equivalent to the built-in C{open} with automatic
file closing behavior.
@param filepath: The path of the file that you want to open
@type filepath: C{str}
@param mode: POSIX mode in which you want to open the file.
@type mode: C{str} see documentation for the format.
@param bufsize: Buffer size for file IO
@type bufsize: C{int} see documentation for the meaning
@param encoding: Encoding that should be used to read the file
@type encoding: C{str}
@param inplace: Indicates if the temporary file should reside
in the same directory of the final file.
@type inplace: C{bool}
@param tmpd: The temporary directory in which file IO is
performed. Then files are moved from here to
their original destination.
@type tmpd: C{str}
@param sync: Force topen to fsync the file before closing it
@type sync: C{bool}
"""
if 'r' in mode or 'a' in mode:
fp = filepath
else:
if inplace:
source_dir, _ = os.path.split(filepath)
tmpd = source_dir
if not os.path.exists(tmpd):
os.makedirs(tmpd)
_f, fp = tempfile.mkstemp(dir=tmpd)
if encoding is not None:
f = codecs.open(fp, mode, encoding=encoding)
else:
f = open(fp, mode, bufsize)
try:
yield f
finally:
if 'r' in mode:
if "+" in mode:
f.flush()
if sync:
os.fsync(f.fileno())
f.close()
return
f.flush()
if sync:
os.fsync(f.fileno())
f.close()
if 'w' in mode:
shutil.move(fp, filepath)
if __name__ == "__main__":
with topen("a_test") as f:
f.write("hello")
assert file("a_test", "rb").read() == 'hello'
assert os.path.exists(TDIR)
os.rmdir(TDIR)
with topen("a_test", mode="rb") as f:
assert f.read() == "hello"
assert not os.path.exists(TDIR)
os.remove("a_test")
--
Valentino Volonghi aka Dialtone
Now running MacOS X 10.5
Home Page: http://www.twisted.it
http://www.adroll.com
PGP.sig
Description: This is a digitally signed message part
_______________________________________________ Python-Dev mailing list Python-Dev@python.org http://mail.python.org/mailman/listinfo/python-dev Unsubscribe: http://mail.python.org/mailman/options/python-dev/archive%40mail-archive.com
