python-dev

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

_______________________________________________
Python-Dev mailing list
[EMAIL PROTECTED]
http://mail.python.org/mailman/listinfo/python-dev
Unsubscribe: 
http://mail.python.org/mailman/options/python-dev/alexiscircle%40gmail.com