Un fragmento simple que implementa la escritura atómica usando Python tempfile
.
with open_atomic('test.txt', 'w') as f:
f.write("huzza")
o incluso la lectura y la escritura hacia y desde el mismo archivo:
with open('test.txt', 'r') as src:
with open_atomic('test.txt', 'w') as dst:
for line in src:
f.write(line)
utilizando dos gestores de contexto simples
import os
import tempfile as tmp
from contextlib import contextmanager
@contextmanager
def tempfile(suffix='', dir=None):
""" Context for temporary file.
Will find a free temporary filename upon entering
and will try to delete the file on leaving, even in case of an exception.
Parameters
----------
suffix : string
optional file suffix
dir : string
optional directory to save temporary file in
"""
tf = tmp.NamedTemporaryFile(delete=False, suffix=suffix, dir=dir)
tf.file.close()
try:
yield tf.name
finally:
try:
os.remove(tf.name)
except OSError as e:
if e.errno == 2:
pass
else:
raise
@contextmanager
def open_atomic(filepath, *args, **kwargs):
""" Open temporary file object that atomically moves to destination upon
exiting.
Allows reading and writing to and from the same filename.
The file will not be moved to destination in case of an exception.
Parameters
----------
filepath : string
the file path to be opened
fsync : bool
whether to force write the file to disk
*args : mixed
Any valid arguments for :code:`open`
**kwargs : mixed
Any valid keyword arguments for :code:`open`
"""
fsync = kwargs.get('fsync', False)
with tempfile(dir=os.path.dirname(os.path.abspath(filepath))) as tmppath:
with open(tmppath, *args, **kwargs) as file:
try:
yield file
finally:
if fsync:
file.flush()
os.fsync(file.fileno())
os.rename(tmppath, filepath)
relacionados con:.. [Y threadsafe escritura de archivos tolerante a errores] (http://stackoverflow.com/questions/12003805/threadsafe-and-fault-tolerant-file-writes) – jfs