Python中怎么創(chuàng)建一個Shell

本篇文章給大家分享的是有關(guān)Python中怎么創(chuàng)建一個Shell，小編覺得挺實用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

基礎(chǔ)知識

讓我們從一個類開始。這是一個簡單的方法，將其初始化參數(shù)保存到局部變量，然后使用subprocess.run對其自身進(jìn)行延遲求值并保存結(jié)果。

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

我們讓它旋轉(zhuǎn)一下：

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

使其看起來更像“命令式”

不用每次我們要自定義命令時都去調(diào)用PipePy。

ls_l = PipePy('ls', '-l')  print(ls_l)

相當(dāng)于

ls = PipePy('ls')  print(ls('-l'))

換句話說，我們要使：

PipePy('ls', '-l')

相當(dāng)于

PipePy('ls')('-l')

值得慶幸的是，我們的類創(chuàng)建了惰性對象這一事實在很大程度上幫助了我們：

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

關(guān)鍵字參數(shù)

如果要向ls傳遞更多參數(shù)，則可能會遇到--sort = size。我們可以輕松地執(zhí)行l(wèi)s（'-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

讓我們來旋轉(zhuǎn)一下：

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

事情開始變得有趣起來。我們的最終目標(biāo)是能夠做到：

ls = PipePy('ls')  grep = PipePy('grep')  print(ls | grep('tags'))  # <<< tags

我們的過程是：

1、讓__init__和__call__方法接受一個僅用于關(guān)鍵字的新_pipe_input關(guān)鍵字參數(shù)，該參數(shù)將保存在self上。

2、在評估期間，如果設(shè)置了_pipe_input，它將作為輸入?yún)?shù)傳遞給subprocess.run。

3、重寫__or__方法以將左操作數(shù)的結(jié)果作為pipe輸入傳遞給右操作數(shù)。

 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

現(xiàn)在我們可以作出如下處理：

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())

現(xiàn)在可以作出如下操作：

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())

現(xiàn)在可以作出如下操作：

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

現(xiàn)在可以作出下面操作：

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實用程序：

在子進(jìn)程中更改工作目錄不會影響當(dāng)前的腳本或python shell。與更改環(huán)境變量相同，以下內(nèi)容不是PipePy的補(bǔ)充，但很不錯：

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

我們的實例是惰性的，這意味著如果我們對它們的結(jié)果感興趣，則將對它們進(jìn)行評估，此后不再進(jìn)行評估。如果我們只是想確保已執(zhí)行該操作怎么辦？例如，假設(shè)我們有以下腳本：

from pipepy import PipePy  tar = PipePy('tar')  tar('-xf', 'some_archive.tar')  print("File extracted")

該腳本實際上不會執(zhí)行任何操作，因為tar調(diào)用實際上并未得到評估。我認(rèn)為一個不錯的慣例是，如果不帶參數(shù)調(diào)用__call__強(qiáng)制求值：

 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

因此在編寫腳本時，如果要確保實際上已調(diào)用命令，則必須用一對括號來調(diào)用它：

 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保留在內(nèi)存中。隨后的評估實際上不會調(diào)用該命令，而只是返回存儲的值。

一種解決方案是通過使用空括號來強(qiáng)制創(chuàng)建副本：

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構(gòu)造函數(shù)返回的實例不應(yīng)該是惰性的，但由__call__調(diào)用返回的實例將是惰性的。

 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)

旋轉(zhuǎn)一下：

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

并且可以預(yù)見的是，使用空調(diào)用的返回值將具有之前的行為：

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

沒關(guān)系 您不會期望d會更新其值。

越來越危險

好吧，ls（'-l'）不錯，但是如果我們像人類一樣簡單地做ls -l，那就太好了。嗯，我有個主意：

class PipePy:      # __init__, etc      def __sub__(left, right):          return left(f"-{right}")

現(xiàn)在可以作如下操作：

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

現(xiàn)在無濟(jì)于事：

import string  for char in string.ascii_letters:      if char in locals():          continue     locals()[char] = char  class PipePy:      # __init__, etc

更危險的事情

用locals（）給了我一個靈感。為什么我們必須一直實例化PipePy？我們無法在路徑中找到所有可執(zhí)行文件，并根據(jù)它們創(chuàng)建PipePy實例嗎？我們當(dāng)然可以！

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)

因此，現(xiàn)在，將我們擁有的所有內(nèi)容都放在一個python文件中，并刪除腳本（這是實際bash腳本的轉(zhuǎn)錄）：

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中怎么創(chuàng)建一個Shell，小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降?。希望你能通過這篇文章學(xué)到更多知識。更多詳情敬請關(guān)注億速云行業(yè)資訊頻道。