Python中怎么创建一个Shell

本篇文章给大家分享的是有关python中怎么创建一个shell，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。

基础知识

让我们从一个类开始。这是一个简单的方法，将其初始化参数保存到局部变量，然后使用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

我们的过程是：

让__init__和__call__方法接受一个仅用于关键字的新_pipe_input关键字参数，该参数将保存在self上。

在评估期间，如果设置了_pipe_input，它将作为输入参数传递给subprocess.run。

重写__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

让我们添加一些简单的东西

真实性：

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

读取/写入文件：

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

迭代

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

表格：

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'}]

普通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中怎么创建一个Shell，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注编程网Python频道。