Source code for debinterface.toolutils

# -*- coding: utf-8 -*-
from __future__ import print_function, with_statement, absolute_import
import os
import stat
import tempfile
from contextlib import contextmanager
import subprocess
try:
    import typing as tp
except ImportError:
    pass


[docs]def safe_subprocess(command_array): # type: (tp.Union[tp.List[str], str]) -> tp.Tuple[bool, str] """Executes shell command. Do not raise Args: command_array (list): ideally an array of string elements, but may be a string. Will be converted to an array of strings Returns: bool, str: True/False (command succeeded), command output """ try: # Ensure all args are strings if isinstance(command_array, list): command_array_str = [str(x) for x in command_array] else: command_array_str = [str(command_array)] return True, subprocess.check_output(command_array_str, stderr=subprocess.STDOUT) except OSError as ex: return False, ex.__str__() except subprocess.CalledProcessError as ex: return False, ex.output
@contextmanager
[docs]def atomic_write(filepath): # type: (str) -> tp.Generator[tp.IO[str], None, None] """ Writeable file object that atomically updates a file (using a temporary file). Args: filepath (str): the file path to be opened """ # Put tmp file to same directory as target file, to allow atomic move realpath = os.path.realpath(filepath) tmppath = os.path.dirname(realpath) with tempfile.NamedTemporaryFile(dir=tmppath, delete=False) as tempf: with open(tempf.name, mode='w+') as tmp: yield tmp tmp.flush() os.fsync(tmp.fileno()) os.rename(tempf.name, realpath) os.chmod(realpath, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH)