My Portfolio

RokcyLinuxでシェルコマンドを実行するpythonプログラムのサンプルコード

登録日	：2024/04/06 09:55
カテゴリ	：Python基礎

subprocessを使ってLinuxサーバでシェルコマンドを実行するためのベースとなるコードを作成した。
これをベースに、Linuxサーバで自動的にコマンドを発行して管理するために活用している。
管理サーバから、クライアントノードのステータスを確認する場合、コマンドのタイムアウト時間を設けておく方がいいところを工夫している。

#!/usr/bin/python3

import subprocess
from subprocess import PIPE
import signal

"""
submit shell command
"""

class ShellCommand(object):
    def __init__(self):
        self.stdout = False
        self.stderr = False
        self.returncode = False
        self.command = False

    def submit_command(self, command):
        self.command = command
        result = subprocess.run(self.command, shell=True, stdout=PIPE, stderr=PIPE)
        self.stdout = result.stdout.decode('utf-8')
        self.stderr = result.stderr.decode('utf-8')
        self.returncode = result.returncode

        if result.returncode != 0:
            raise Exception(self.stderr)


class ExecuteShellCommand(object):
    def __init__(self, timeout, command):
        self._timeout = timeout
        self._command = command
        self._command_result = False
        self._errlog = False

    def _signal_handler(self, signum, frame):
        raise Exception({"Time out! ": self._command})

    def execute_command(self):
        signal.signal(signal.SIGALRM, self._signal_handler)
        signal.alarm(self._timeout)

        shell_command = ShellCommand()

        try:
            shell_command.submit_command(self._command)
            self._command_result = shell_command.stdout
        except Exception as e:
            self._errlog = {'[ERROR]:':str(e)}
            print(self._errlog)

        del shell_command


"""
test code
"""
def test_shell_command():

    timeout = 2
    command = 'ls -l'
    shellcommand = ExecuteShellCommand(timeout, command)
    shellcommand.execute_command()
    print(shellcommand._command_result)

if __name__ == '__main__':

    test_shell_command()

少々、クラスの分離がいまいちだったので、以下のように少し見直した。shellcommadのオブジェクトを引数で受け取るようにしている。

#!/usr/bin/python3

from abc import ABC, abstractmethod
import subprocess
from subprocess import PIPE
import signal

"""
submit shell command
"""

class ShellCommand(object):
    def __init__(self):
        self.stdout = False
        self.stderr = False
        self.returncode = False
        self.command = False

    def submit_command(self, command):
        self.command = command
        result = subprocess.run(self.command, shell=True, stdout=PIPE, stderr=PIPE)
        self.stdout = result.stdout.decode('utf-8')
        self.stderr = result.stderr.decode('utf-8')
        self.returncode = result.returncode

        if result.returncode != 0:
            raise Exception(self.stderr)


class ExecuteShellCommand(object):
    def __init__(self, timeout: int, command: str, shellcommand: ShellCommand):
        self._timeout = timeout
        self._command = command
        self._shellcommand = shellcommand
        self._command_result = False
        self._errlog = False

    def _signal_handler(self, signum, frame):
        raise Exception({"Time out! ": self._command})

    def execute_command(self):
        signal.signal(signal.SIGALRM, self._signal_handler)
        signal.alarm(self._timeout)

        try:
            self._shellcommand.submit_command(self._command)
            self._command_result = self._shellcommand.stdout
        except Exception as e:
            self._errlog = {'[ERROR]:':str(e)}
            print(self._errlog)


"""
test code
"""
def test_shell_command():

    timeout = 2
    command = 'ls -l'
    shellcommand = ShellCommand()
    shell = ExecuteShellCommand(timeout, command, shellcommand)
    shell.execute_command()
    print(shell._command_result)


if __name__ == '__main__':

    test_shell_command()