def _run_cmd()

in mysqloperator/controller/backup/meb/meb_controller.py [0:0]

• 42 lines of code
• 10 McCabe index (conditional complexity)

    def _run_cmd(self, cmd: str, args=None):
        command = [
            MYSQLBACKUP_BINARY,
            '--backup-dir='+self.tmppath
        ]

        if args:
            command += args

        if cmd != "copy-back-and-apply-log":
            command += self.storage.get_args() + self.auth_options

        command.append(cmd)

        def print_output(pipe, stream):
            hidden = self.storage.get_filter()

            try:
                for line in iter(pipe.readline, ''):
                    if line:
                        if hidden:
                            # Avoid printing PAR URLs and similar into logs
                            for term in hidden:
                                line = line.replace(term, 'X' * len(term))

                        self.log += line

                        stream.write(line)
                        stream.flush()
            finally:
                pipe.close()

        # in order to pass password via stdin we got to also pipe stdout/stderr
        # through our code, if we end up using localroot (passwordless) we
        # might use the simpler subprocess.run ...
        process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=1,
            text=True
        )

        process.stdin.close()

        stdout_thread = threading.Thread(target=print_output, args=(process.stdout, sys.stdout))
        stderr_thread = threading.Thread(target=print_output, args=(process.stderr, sys.stderr))

        stdout_thread.start()
        stderr_thread.start()

        stdout_thread.join()
        stderr_thread.join()

        error_code = process.wait()

        if error_code:
            raise RuntimeError(f"Running Backup failed: {error_code}")

        return True