Skip to content

Instantly share code, notes, and snippets.

@samuelsh
Created August 3, 2016 06:52
Show Gist options
  • Save samuelsh/f7eed6460fd9cc4e884218df4b53bec3 to your computer and use it in GitHub Desktop.
Save samuelsh/f7eed6460fd9cc4e884218df4b53bec3 to your computer and use it in GitHub Desktop.
Set of shell utils that allows to run shell commands, shell scripts and bash functions on local and remote hosts
class ShellUtils:
def __init__(self):
pass
@staticmethod
def run_bash_function(library_path, function_name, params):
cmdline = ['bash', '-c', '. %s; %s %s' % (library_path, function_name, params)]
p = subprocess.Popen(cmdline,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
raise RuntimeError("%r failed, status code %s stdout %r stderr %r" % (
function_name, p.returncode, stdout, stderr))
return stdout.strip() # This is the stdout from the shell command
@staticmethod
def run_remote_bash_function(remote_host, library_path, function_name, params):
cmdline = ['ssh', '-nx', remote_host, 'bash', '-c', '. %s; %s %s' % (library_path, function_name, params)]
p = subprocess.Popen(cmdline,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
raise RuntimeError("%r failed, status code %s stdout %r stderr %r" % (
function_name, p.returncode, stdout, stderr))
return stdout.strip() # This is the stdout from the shell command
@staticmethod
def run_shell_script(script, params, stdout=True):
FNULL = open(os.devnull, 'w')
if not stdout:
p = subprocess.call([script, params], stdout=FNULL)
else:
p = subprocess.call([script, params])
return p
@staticmethod
def run_shell_script_remote(remote_host, script, params, stdout=True):
FNULL = open(os.devnull, 'w')
if not stdout:
p = subprocess.call(['ssh', '-nx', remote_host, script, params], stdout=FNULL)
else:
p = subprocess.call(['ssh', '-nx', remote_host, script, params])
return p
@staticmethod
def run_shell_command(cmd, params, stdout=subprocess.PIPE):
cmdline = [cmd]
cmdline = cmdline + params.split(' ')
p = subprocess.Popen(cmdline, stdout=stdout, stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
raise RuntimeError("%r failed, status code %s stdout %r stderr %r" % (
cmd, p.returncode, stdout, stderr))
return stdout.strip() # This is the stdout from the shell command
@staticmethod
def run_shell_remote_command(remote_host, remote_cmd):
p = subprocess.Popen([SSH_PATH, '-nx', remote_host, remote_cmd], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
raise RuntimeError("%r failed, status code %s stdout %r stderr %r" % (
remote_cmd, p.returncode, stdout, stderr))
return stdout.strip() # This is the stdout from the shell command
@staticmethod
def run_shell_remote_command_multiline(remote_host, remote_cmd):
p = subprocess.Popen(['ssh', '-nx', remote_host, remote_cmd], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
raise RuntimeError("%r failed, status code %s stdout %r stderr %r" % (
remote_cmd, p.returncode, stdout, stderr))
return stdout.splitlines() # This is the stdout from the shell command
@staticmethod
def run_shell_remote_command_background(remote_host, remote_cmd):
subprocess.Popen(['ssh', '-nx', remote_host, remote_cmd])
@staticmethod
def run_shell_remote_command_no_exception(remote_host, remote_cmd):
p = subprocess.Popen(['ssh', '-nx', remote_host, remote_cmd], stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = p.communicate()
if p.returncode != 0:
return False # This is the stdout from the shell command
return True
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment