Revert "utils: add an atomic_write() context manager (#32413)"

This reverts commit a52c914e8f.
This commit is contained in:
Benjamin Dauvergne 2019-04-19 11:19:11 +02:00
parent 1a60c1acd3
commit 72489a1707
2 changed files with 0 additions and 80 deletions

View File

@ -1,42 +0,0 @@
# Copyright (C) 2018 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os.path
import contextlib
import tempfile
@contextlib.contextmanager
def atomic_write(filepath, **kwargs):
'''Return a file descriptor to a temporary file using NamedTemporaryFile
which will be atomically renamed to filepath if possible.
Atomic renaming is only possible on the same filesystem, so the
temporary file will be created in the same directory as the target file
You can pass any possible argument to NamedTemporaryFile with kwargs.
'''
target_dir = kwargs.get('dir') or os.path.dirname(filepath)
fd = tempfile.NamedTemporaryFile(dir=target_dir, delete=False)
try:
with fd:
yield fd
fd.flush()
os.fsync(fd.fileno())
os.rename(fd.name, filepath)
except Exception:
os.unlink(fd.name)
raise

View File

@ -1,38 +0,0 @@
# Copyright (C) 2018 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import pytest
from passerelle.utils.files import atomic_write
def test_atomic_write(tmpdir):
filepath = str(tmpdir.join('test'))
with pytest.raises(Exception):
with atomic_write(filepath) as fd:
fd.write('coucou')
raise Exception()
assert not os.path.exists(filepath)
assert os.listdir(str(tmpdir)) == []
with atomic_write(filepath) as fd:
fd.write('coucou')
assert os.path.exists(filepath)
with open(filepath) as fd:
assert fd.read() == 'coucou'