This catches the OSError thrown by the subprocess module and wraps it so that in the end only an error message is printed, explaining which command failed.
72 lines
1.6 KiB
Python
72 lines
1.6 KiB
Python
import contextlib, subprocess, tempfile, shutil, re, os
|
|
|
|
|
|
class UserError(Exception):
|
|
def __init__(self, message, *args):
|
|
super(UserError, self).__init__(message.format(*args))
|
|
|
|
|
|
def rename_atomic(source_path, target_path):
|
|
"""
|
|
Move the file at source_path to target_path.
|
|
|
|
If both paths reside on the same device, os.rename() is used, otherwise the file is copied to a temporary name next to target_path and moved from there using os.rename().
|
|
"""
|
|
|
|
source_dir_stat = os.stat(os.path.dirname(source_path))
|
|
target_dir_stat = os.stat(os.path.dirname(target_path))
|
|
|
|
if source_dir_stat.st_dev == target_dir_stat.st_dev:
|
|
os.rename(source_path, target_path)
|
|
else:
|
|
temp_path = target_path + '~'
|
|
|
|
shutil.copyfile(source_path, temp_path)
|
|
os.rename(temp_path, target_path)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def TemporaryDirectory():
|
|
dir = tempfile.mkdtemp()
|
|
|
|
try:
|
|
yield dir
|
|
finally:
|
|
shutil.rmtree(dir)
|
|
|
|
|
|
def command(args, remove_env = [], set_env = { }):
|
|
env = dict(os.environ)
|
|
|
|
for i in remove_env:
|
|
del env[i]
|
|
|
|
for k, v in set_env.items():
|
|
env[k] = v
|
|
|
|
try:
|
|
process = subprocess.Popen(args, env = env)
|
|
process.wait()
|
|
except OSError as e:
|
|
raise UserError('Error running {}: {}', args[0], e)
|
|
|
|
if process.returncode:
|
|
raise UserError('Command failed: {}', ' '.join(args))
|
|
|
|
|
|
def bash_escape_string(string):
|
|
return "'{}'".format(re.sub("'", "'\"'\"'", string))
|
|
|
|
|
|
def write_file(path, data):
|
|
temp_path = path + '~'
|
|
|
|
with open(temp_path, 'wb') as file:
|
|
file.write(data)
|
|
|
|
os.rename(temp_path, path)
|
|
|
|
|
|
def read_file(path):
|
|
with open(path, 'rb') as file:
|
|
return file.read()
|