用Python創建你自己的Shell
作者：佚名
我們不能否認將整個程序當作純粹的功能發揮作用的能力，以及將一個程序的輸出傳遞到另一個程序的自然程度。因此，我想知道，我們能否將bash的某些功能與Python結合起來。
介紹
很多人討厭bash腳本。每當我要做最簡單的事情時，我都必須查閱文檔。如何將函數的參數轉發給子命令？如何將字符串分配給變量，然后作為命令調用該字符串？如何檢查兩個字符串變量是否相等？如何分割字符串并獲得后半部分？等等。不是我找不到這些答案，而是每次都必須查找它們。
但是，我們不能否認將整個程序當作純粹的功能發揮作用的能力，以及將一個程序的輸出傳遞到另一個程序的自然程度。因此，我想知道，我們能否將bash的某些功能與Python結合起來？
基礎知識
讓我們從一個類開始。這是一個簡單的方法，將其初始化參數保存到局部變量，然后使用subprocess.run對其自身進行延遲求值并保存結果。
- import subprocess
- class PipePy:
- def __init__(self, *args):
- self._args = args
- self._result = None
- def _evaluate(self):
- if self._result is not None:
- return
- self._result = subprocess.run(self._args,
- capture_output=True,
- text=True)
- @property
- def returncode(self):
- self._evaluate()
- return self._result.returncode
- @property
- def stdout(self):
- self._evaluate()
- return self._result.stdout
- def __str__(self):
- return self.stdout
- @property
- def stderr(self):
- self._evaluate()
- return self._result.stderr
我們讓它旋轉一下：
- ls = PipePy('ls')
- ls_l = PipePy('ls', '-l')
- print(ls)
- # <<< files.txt
- # ... main.py
- # ... tags
- print(ls_l)
- # <<< total 16
- # ... -rw-r--r-- 1 kbairak kbairak 125 Jan 22 08:53 files.txt
- # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb 1 21:54 main.py
- # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
使其看起來更像“命令式”
不用每次我們要自定義命令時都去調用PipePy。
- ls_l = PipePy('ls', '-l')
- print(ls_l)
相當于
- ls = PipePy('ls')
- print(ls('-l'))
換句話說，我們要使：
- PipePy('ls', '-l')
相當于
- PipePy('ls')('-l')
值得慶幸的是，我們的類創建了惰性對象這一事實在很大程度上幫助了我們：
- class PipePy:
- # __init__, etc
- def __call__(self, *args):
- args = self._args + args
- return self.__class__(*args)
- ls = PipePy('ls')
- print(ls('-l'))
- # <<< total 16
- # ... -rw-r--r-- 1 kbairak kbairak 125 Jan 22 08:53 files.txt
- # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb 1 21:54 main.py
- # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
關鍵字參數
如果要向ls傳遞更多參數，則可能會遇到--sort = size。我們可以輕松地執行ls（'-l'，'--sort = size'）。我們可以做得更好嗎？
- class PipePy:
- - def __init__(self, *args):
- + def __init__(self, *args, **kwargs):
- self._args = args
- + self._kwargs = kwargs
- self._result = None
- def _evaluate(self):
- if self._result is not None:
- return
- - self._result = subprocess.run(self._args,
- + self._result = subprocess.run(self._convert_args(),
- capture_output=True,
- text=True)
- + def _convert_args(self):
- + args = [str(arg) for arg in self._args]
- + for key, value in self._kwargs.items():
- + keykey = key.replace('_', '-')
- + args.append(f"--{key}={value}")
- + return args
- - def __call__(self, *args):
- + def __call__(self, *args, **kwargs):
- args = self._args + args
- + kwargs = {**self._kwargs, **kwargs}
- - return self.__class__(*args)
- + return self.__class__(*args, **kwargs)
- # returncode, etc
讓我們來旋轉一下：
- print(ls('-l'))
- # <<< total 16
- # ... -rw-r--r-- 1 kbairak kbairak 125 Jan 22 08:53 files.txt
- # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb 1 21:54 main.py
- # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
- print(ls('-l', sort="size"))
- # <<< total 16
- # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb 1 21:54 main.py
- # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
- # ... -rw-r--r-- 1 kbairak kbairak 125 Jan 22 08:53 files.txt
Piping
事情開始變得有趣起來。我們的最終目標是能夠做到：
- ls = PipePy('ls')
- grep = PipePy('grep')
- print(ls | grep('tags'))
- # <<< tags
我們的過程是：
1、讓__init__和__call__方法接受一個僅用于關鍵字的新_pipe_input關鍵字參數，該參數將保存在self上。
2、在評估期間，如果設置了_pipe_input，它將作為輸入參數傳遞給subprocess.run。
3、重寫__or__方法以將左操作數的結果作為pipe輸入傳遞給右操作數。
- class PipePy:
- - def __init__(self, *args, **kwargs):
- + def __init__(self, *args, _pipe_input=None, **kwargs):
- self._args = args
- self._kwargs = kwargs
- + self._pipe_input = _pipe_input
- self._result = None
- - def __call__(self, *args, **kwargs):
- + def __call__(self, *args, _pipe_input=None, **kwargs):
- args = self._args + args
- kwargs = {**self._kwargs, **kwargs}
- - return self.__class__(*args, **kwargs)
- + return self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)
- def _evaluate(self):
- if self._result is not None:
- return
- self._result = subprocess.run(self._convert_args(),
- + input=self._pipe_input,
- capture_output=True,
- text=True)
- + def __or__(left, right):
- + return right(_pipe_input=left.stdout)
讓我們嘗試一下（從之前稍微修改命令以證明它確實有效）：
- ls = PipePy('ls')
- grep = PipePy('grep')
- print(ls('-l') | grep('tags'))
- # <<< -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
讓我們添加一些簡單的東西
1、真實性：
- class PipePy:
- # __init__, etc
- def __bool__(self):
- return self.returncode == 0
現在我們可以作出如下處理：
- git = PipePy('git')
- grep = PipePy('grep')
- if git('branch') | grep('my_feature'):
- print("Branch 'my_feature' found")
2、讀取/寫入文件：
- class PipePy:
- # __init__, etc
- def __gt__(self, filename):
- with open(filename, 'w') as f:
- f.write(self.stdout)
- def __rshift__(self, filename):
- with open(filename, 'a') as f:
- f.write(self.stdout)
- def __lt__(self, filename):
- with open(filename) as f:
- return self(_pipe_input=f.read())
現在可以作出如下操作：
- ls = PipePy('ls')
- grep = PipePy('grep')
- cat = PipePy('cat')
- ls > 'files.txt'
- print(grep('main') < 'files.txt')
- # <<< main.py
- ls >> 'files.txt'
- print(cat('files.txt'))
- # <<< files.txt
- # ... main.py
- # ... tags
- # ... files.txt
- # ... main.py
- # ... tags
3、迭代
- class PipePy:
- # __init__, etc
- def __iter__(self):
- return iter(self.stdout.split())
現在可以作出如下操作：
- ls = PipePy('ls')
- for name in ls:
- print(name.upper())
- # <<< FILES.TXT
- # ... MAIN.PY
- # ... TAGS
4、表格：
- class PipePy:
- # __init__, etc
- def as_table(self):
- lines = self.stdout.splitlines()
- fields = lines[0].split()
- result = []
- for line in lines[1:]:
- item = {}
- for i, value in enumerate(line.split(maxsplit=len(fields) - 1)):
- item[fields[i]] = value
- result.append(item)
- return result
現在可以作出下面操作：
- ps = PipePy('ps')
- print(ps)
- # <<< PID TTY TIME CMD
- # ... 4205 pts/4 00:00:00 zsh
- # ... 13592 pts/4 00:00:22 ptipython
- # ... 16253 pts/4 00:00:00 ps
- ps.as_table()
- # <<< [{'PID': '4205', 'TTY': 'pts/4', 'TIME': '00:00:00', 'CMD': 'zsh'},
- # ... {'PID': '13592', 'TTY': 'pts/4', 'TIME': '00:00:22', 'CMD': 'ptipython'},
- # ... {'PID': '16208', 'TTY': 'pts/4', 'TIME': '00:00:00', 'CMD': 'ps'}]
5、普通bash實用程序：
在子進程中更改工作目錄不會影響當前的腳本或python shell。與更改環境變量相同，以下內容不是PipePy的補充，但很不錯：
- import os
- cd = os.chdir
- export = os.environ.__setitem__
- pwd = PipePy('pwd')
- pwd
- # <<< /home/kbairak/prog/python/pipepy
- cd('..')
- pwd
- # <<< /home/kbairak/prog/python
使事情看起來更shell-like
如果我在交互式shell中，則希望能夠簡單地鍵入ls并完成它。
- class PipePy:
- # __init__, etc
- def __repr__(self):
- return self.stdout + self.stderr
交互式shell
- >>> ls = PipePy('ls')
- >>> ls
- files.txt
- main.py
- tags
我們的實例是惰性的，這意味著如果我們對它們的結果感興趣，則將對它們進行評估，此后不再進行評估。如果我們只是想確保已執行該操作怎么辦？例如，假設我們有以下腳本：
- from pipepy import PipePy
- tar = PipePy('tar')
- tar('-xf', 'some_archive.tar')
- print("File extracted")
該腳本實際上不會執行任何操作，因為tar調用實際上并未得到評估。我認為一個不錯的慣例是，如果不帶參數調用__call__強制求值：
- class PipePy:
- def __call__(self, *args, _pipe_input=None, **kwargs):
- args = self._args + args
- kwargs = {**self._kwargs, **kwargs}
- - return self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)
- + result = self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)
- + if not args and not _pipe_input and not kwargs:
- + result._evaluate()
- + return result
因此在編寫腳本時，如果要確保實際上已調用命令，則必須用一對括號來調用它：
- from pipepy import PipePy
- tar = PipePy('tar')
- -tar('-xf', 'some_archive.tar')
- +tar('-xf', 'some_archive.tar')()
- print("File extracted")
但是，我們還沒有解決問題?？紤]一下：
- date = PipePy('date')
- date
- # <<< Mon Feb 1 10:43:08 PM EET 2021
- # Wait 5 seconds
- date
- # <<< Mon Feb 1 10:43:08 PM EET 2021
不好！date沒有改變。date對象將其_result保留在內存中。隨后的評估實際上不會調用該命令，而只是返回存儲的值。
一種解決方案是通過使用空括號來強制創建副本：
- date = PipePy('date')
- date()
- # <<< Mon Feb 1 10:45:09 PM EET 2021
- # Wait 5 seconds
- date()
- # <<< Mon Feb 1 10:45:14 PM EET 2021
另一個解決方案是：由PipePy構造函數返回的實例不應該是惰性的，但由__call__調用返回的實例將是惰性的。
- class PipePy:
- - def __init__(self, *args, _pipe_input=None, **kwargs):
- + def __init__(self, *args, _pipe_input=None, _lazy=False, **kwargs):
- self._args = args
- self._kwargs = kwargs
- self._pipe_input = _pipe_input
- + self._lazy = _lazy
- self._result = None
- def __call__(self, *args, _pipe_input=None, **kwargs):
- args = self._args + args
- kwargs = {**self._kwargs, **kwargs}
- - result = self.__class__(*args, _pipe_input_pipe_input=_pipe_input, **kwargs)
- + result = self.__class__(*args,
- + _pipe_input_pipe_input=_pipe_input,
- + _lazy=True,
- + **kwargs)
- if not args and not _pipe_input and not kwargs:
- result._evaluate()
- return result
- def _evaluate(self):
- - if self._result is not None:
- + if self._result is not None and self._lazy:
- return
- self._result = subprocess.run(self._convert_args(),
- input=self._pipe_input,
- capture_output=True,
- text=True)
旋轉一下：
- date = PipePy('date')
- date
- # <<< Mon Feb 1 10:54:09 PM EET 2021
- # Wait 5 seconds
- date
- # <<< Mon Feb 1 10:54:14 PM EET 2021
并且可以預見的是，使用空調用的返回值將具有之前的行為：
- date = PipePy('date')
- d = date()
- d
- # <<< Mon Feb 1 10:56:21 PM EET 2021
- # Wait 5 seconds
- d
- # <<< Mon Feb 1 10:56:21 PM EET 2021
沒關系 您不會期望d會更新其值。
越來越危險
好吧，ls（'-l'）不錯，但是如果我們像人類一樣簡單地做ls -l，那就太好了。嗯，我有個主意：
- class PipePy:
- # __init__, etc
- def __sub__(left, right):
- return left(f"-{right}")
現在可以作如下操作：
- ls = PipePy('ls')
- ls - 'l'
- # <<< total 16
- # ... -rw-r--r-- 1 kbairak kbairak 46 Feb 1 23:04 files.txt
- # ... -rw-r--r-- 1 kbairak kbairak 5425 Feb 1 21:54 main.py
- # ... -rw-r--r-- 1 kbairak kbairak 1838 Feb 1 21:54 tags
我們還有一步：
- l = 'l'
- ls -l
現在無濟于事：
- import string
- for char in string.ascii_letters:
- if char in locals():
- continue
- locals()[char] = char
- class PipePy:
- # __init__, etc
更危險的事情
用locals（）給了我一個靈感。為什么我們必須一直實例化PipePy？我們無法在路徑中找到所有可執行文件，并根據它們創建PipePy實例嗎？我們當然可以！
- import os
- import stat
- for path in os.get_exec_path():
- try:
- names = os.listdir(path)
- except FileNotFoundError:
- continue
- for name in names:
- if name in locals():
- continue
- if 'x' in stat.filemode(os.lstat(os.path.join(path, name)).st_mode):
- locals()[name] = PipePy(name)
因此，現在，將我們擁有的所有內容都放在一個python文件中，并刪除腳本（這是實際bash腳本的轉錄）：
- from pipepy import mysqladmin, sleep, drush, grep
- for i in range(10):
- if mysqladmin('ping',
- host="mysql_drupal7",
- user="user",
- password="password"):
- break
- sleep(1)() # Remember to actually invoke
- if not drush('status', 'bootstrap') | grep('-q', 'Successful'):
- drush('-y', 'site-install', 'standard',
- db_url="mysql://user:password@mysql_drupal7:3306/drupal",
- acount_pass="kbairak")() # Remember to actually invoke
- drush('en', 'tmgmt_ui', 'tmgmt_entity_ui', 'tmgmt_node_ui')()
責任編輯：龐桂玉
來源：
Python中文社區 (ID:python-china)
