用 Python 创建你自己的 Shell

介绍

很多人讨厌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)
# <<
# ... main.py
# ... tags
print(ls_l)
# <<
# ... -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'))
# <<
# ... -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():
+            key = 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'))
# <<
# ... -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"))
# <<
# ... -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'))
# <<

我们的过程是：

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, **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'))
# <<

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

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')
# <<

ls >> 'files.txt'
print(cat('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())
# <<
# ... 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)
# <<
# ...    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': '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
# <<

cd('..')
pwd
# <<

使事情看起来更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, **kwargs)
+        result = self.__class__(*args, _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
# <<

# Wait 5 seconds

date
# <<

不好！date没有改变。date对象将其_result保留在内存中。随后的评估实际上不会调用该命令，而只是返回存储的值。

一种解决方案是通过使用空括号来强制创建副本：

date = PipePy('date')
date()
# <<

# Wait 5 seconds

date()
# <<

另一个解决方案是：由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, **kwargs)
+        result = self.__class__(*args,
+                                _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
# <<

# Wait 5 seconds

date



    
# <<

并且可以预见的是，使用空调用的返回值将具有之前的行为：

date = PipePy('date')
d = date()
d
# <<

# Wait 5 seconds

d
# <<

没关系您不会期望d会更新其值。

越来越危险

好吧，ls（'-l'）不错，但是如果我们像人类一样简单地做ls -l，那就太好了。嗯，我有个主意：

class PipePy:
    # __init__, etc

    def __sub__(left, right):
        return left(f"-{right}")

现在可以作如下操作：

ls = PipePy('ls')
ls - 'l'
# <<
# ... -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')()

更多阅读

2020 年最佳流行 Python 库 Top 10

2020 Python中文社区热门文章 Top 10

5分钟快速掌握 Python 定时任务框架

特别推荐

点击下方阅读原文加入社区会员