本篇文章给大家分享的是有关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.txtPiping
事情开始变得有趣起来。我们的最终目标是能够做到：
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 # ... tags3、迭代
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 # ... TAGS4、表格：
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中怎么创建一个Shell，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
