Mercurial > hg > AtomicWrite
changeset 1:a4188f41ca35 default tip
basically the thing
author | Jeff Hammel <k0scist@gmail.com> |
---|---|
date | Fri, 02 Jan 2015 13:22:09 -0800 |
parents | dc90512b9c98 |
children | |
files | atomic/writer.py tests/test_atomicwrite.py |
diffstat | 2 files changed, 37 insertions(+), 1 deletions(-) [+] |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/atomic/writer.py Fri Jan 02 13:22:09 2015 -0800 @@ -0,0 +1,33 @@ +# -*- coding: utf-8 -*- + +""" +write a file atomically +""" + +# imports +import os +import shutil +import tempfile + +# module globals +__all__ = ['ensure_dir', 'write'] + +def ensure_dir(directory): + """ensure a directory exists""" + if os.path.exists(directory): + if not os.path.isdir(directory): + raise OSError("Not a directory: '{}'".format(directory)) + return directory + os.makedirs(directory) + return directory + + +def write(contents, path): + """atomically write a file taking advantage of move""" + + path = os.path.abspath(path) + fd, tmp_path = tempfile.mkstemp() + os.write(fd, contents) + os.close(fd) + ensure_dir(os.path.dirname(path)) + shutil.move(tmp_path, path)
--- a/tests/test_atomicwrite.py Fri Jan 02 13:08:39 2015 -0800 +++ b/tests/test_atomicwrite.py Fri Jan 02 13:22:09 2015 -0800 @@ -4,8 +4,8 @@ unit tests for AtomicWrite """ +import atomic import os -import sys import tempfile import unittest @@ -14,6 +14,9 @@ def test_atomicwrite(self): tf = tempfile.mktemp() self.assertFalse(os.path.exists(tf)) + atomic.write('foo', tf) + self.assertTrue(os.path.exists(tf)) + self.assertEqual(open(tf, 'r').read(), 'foo') if __name__ == '__main__': unittest.main()