pipepy · PyPI · Python 包索引

一个用于调用和交互shell命令的Python库

这些详情尚未由PyPI 验证

项目链接

主页

项目描述

一个用于调用和交互shell命令的Python库。

为什么？与其他类似框架的比较

Xonsh：Xonsh允许您将shell和Python结合使用，并实现强大的脚本和交互式会话。这个库在一定程度上也做到了这一点。然而，Xonsh引入了一种新的语言，它是Python的超集。这个库的主要目标是成为一个纯Python实现，主要面向脚本。
sh 和 pieshell：它们与当前库非常相似，因为它们是纯Python实现。然而，当前库试图在以下方面进行改进
- 它试图应用更多的语法糖，使调用感觉更像shell调用。
- 它试图提供方法，让shell命令以强大且直观的方式与Python代码交互。

安装和测试

python -m pip install pipepy

或者，如果您想在尝试代码的同时修改它

git clone https://github.com/kbairak/pipepy
cd pipepy
python -m pip install  -e .

要运行测试，您需要首先安装测试需求

python -m pip install -r test_requirements.txt

pymake test
# or
pytest

在开发过程中，还有一些其他pymake目标可以帮助进行测试

covtest：生成并打开覆盖率报告
watchtest：监听源代码文件的更改并自动重新运行测试
debugtest：运行测试而不捕获它们的输出，以便您可以插入调试语句

pymake 是一个控制台脚本，它是 pipepy 的一部分，旨在替代 GNU make，区别在于 Makefile 是用 Python 编写的。有关更多信息，请参见下文下面。

简介，基本用法

from pipepy import ls, grep

print(ls)  # prints contents of current folder
if ls | grep('info.txt'):
      print('info.txt found')

大多数 shell 命令都可以直接从 pipepy 模块导入。命令名称中的连字符被转换为下划线（docker-compose → docker_compose）。无法自动找到的命令可以使用 PipePy 构造函数创建

from pipepy import PipePy

custom_command = PipePy('./bin/custom')
python_script = PipePy('python', 'script.py')

自定义命令

使用非空参数调用命令将返回一个修改后的未评估副本。因此以下都是等效的

from pipepy import PipePy
ls_l = PipePy('ls', '-l')
# Is equivalent to
ls_l = PipePy('ls')('-l')

您还有许多其他方法可以自定义命令

通配符：通配符将应用于所有位置参数

from pipepy import echo
print(echo('*'))  # Will print all files in the current folder

如果您想避免此功能，可以使用 glob.escape

import glob
from pipepy import ls, echo

print(ls)
# <<< **a *a *aa

print(echo('*a'))
# <<< **a *a *aa

print(echo(glob.escape('*a')))
# <<< *a

关键字参数:

from pipepy import ls
ls(sort="size")     # Equivalent to ls('--sort=size')
ls(I="files.txt")   # Equivalent to ls('-I', 'files.txt')
ls(sort_by="size")  # Equivalent to ls('--sort-by=size')
ls(escape=True)     # Equivalent to ls('--escape')
ls(escape=False)    # Equivalent to ls('--no-escape')

由于关键字参数位于位置参数之后，如果您想使最终命令具有不同的排序顺序，则可以多次调用该命令

from pipepy import ls
ls('-l', sort="size")  # Equivalent to ls('-l', '--sort=size')
ls(sort="size")('-l')  # Equivalent to ls('--sort=size', '-l')

属性访问:

from pipepy import git
git.push.origin.bugfixes  # Equivalent to git('push', 'origin', 'bugfixes')

减号:

from pipepy import ls
ls - 'l'        # Equivalent to ls('-l')
ls - 'default'  # Equivalent to ls('--default')

这是为了使调用看起来更像 shell

from pipepy import ls
l, t = 'l', 't'
ls -l -t  # Equivalent to ls('-l', '-t')

您可以在脚本中调用 pipepy.overload_chars(locals()) 以将所有 ASCII 字母分配给同名的变量。

import pipepy
from pipepy import ls
pipepy.overload_chars(locals())
ls -l -t  # Equivalent to ls('-l', '-t')

懒惰

命令是延迟评估的。例如，这实际上不会做任何事情

from pipepy import wget
wget('http://...')

使用非空参数调用 PipePy 实例将返回一个未评估的副本，该副本提供了额外的参数。当使用输出时，将评估命令。这可以通过以下方式完成

访问 returncode、stdout 和 stderr 属性

from pipepy import echo
command = echo("hello world")
command.returncode
# <<< 0
command.stdout
# <<< 'hello world\n'
command.stderr
# <<< ''

将命令评估为字符串对象
```
from pipepy import ls
result = str(ls)
# or
print(ls)
```
将命令转换为 str 返回其 stdout。

将命令评估为布尔对象

from pipepy import ls, grep
command = ls | grep('info.txt')

bool(command)
# <<< True

if command:
    print("info.txt found")

如果命令的 returncode 为 0，则该命令为真。

调用 .as_table() 方法

from pipepy import ps
ps.as_table()
# <<< [{'PID': '11233', 'TTY': 'pts/4', 'TIME': '00:00:01', 'CMD': 'zsh'},
# ...  {'PID': '17673', 'TTY': 'pts/4', 'TIME': '00:00:08', 'CMD': 'ptipython'},
# ...  {'PID': '18281', 'TTY': 'pts/4', 'TIME': '00:00:00', 'CMD': 'ps'}]

遍历命令对象

from pipepy import ls
for filename in ls:
    print(filename.upper)

command.iter_words() 遍历命令的 stdout 中的单词

from pipepy import ps
list(ps.iter_words())
# <<< ['PID', 'TTY', 'TIME', 'CMD',
# ...  '11439', 'pts/5', '00:00:00', 'zsh',
# ...  '15532', 'pts/5', '00:00:10', 'ptipython',
# ...  '15539', 'pts/5', '00:00:00', 'ps']

将输出重定向到其他位置（这将在下文进一步解释）

from pipepy import ls, grep
ls > 'files.txt'
ls >> 'files.txt'
ls | grep('info.txt')  # `ls` will be evaluated, `grep` will not
ls | lambda output: output.upper()

如果您对命令的输出不感兴趣，但仍然想评估它，则可以使用空参数调用它。因此，实际上将调用该命令（并等待其完成）。

from pipepy import wget
wget('http://...')()

后台命令

在 PipePy 实例上调用 .delay() 将返回一个副本，尽管没有评估，但它将开始在后台运行（从 Celery 的 .delay() 方法中汲取灵感）。再次提醒，如果您尝试访问其输出，它将执行其余的评估过程，这仅仅是等待它完成。

from pipepy import wget
urls = [...]

# All downloads will happen in the background simultaneously
downloads = [wget(url).delay() for url in urls]

# You can do something else here in Python while the downloads are working

# This will call __bool__ on all downloads and thus wait for them
if not all(downloads):
   print("Some downloads failed")

如果您对后台命令的输出不感兴趣，您应该在某个时候小心地调用它上的 .wait()。否则，其进程将不会被等待，如果父 Python 进程结束，它将杀死所有后台进程

from pipepy import wget
download = wget('...').delay()
# Do something else
download.wait()

您可以为 wait 提供可选的 timeout 参数。如果设置了超时，并且进程没有完成，将引发 TimeoutExpired 异常。（这是与 subprocess 模块相同的 TimeoutExpired 异常类，但您也可以从 pipepy 模块导入它）

from pipepy import sleep
command = sleep(100).delay()
command.wait(5)
# <<< TimeoutExpired: Command '['sleep', '30']' timed out after 5 seconds

在任何时候，您都可以调用 pipepy.jobs() 来获取未等待的命令列表。如果您想进行一些清理，还有一个 pipepy.wait_jobs() 函数。但是请注意，如果任何后台作业没有完成或卡住，wait_jobs() 可能会挂起未知的时间。 wait_jobs 也接受可选的 timeout 参数。

重定向文件输出

《>》、《>>》和《<》运算符在shell中的工作方式与在Python中类似

ls               >  'files.txt'  # Will overwrite files.txt
ls               >> 'files.txt'  # Will append to files.txt
grep('info.txt') <  'files.txt'  # Will use files.txt as input

它们也适用于文件类对象

import os
from pipepy import ls, grep

buf = io.StringIO()
ls > buf
ls('subfolder') >> buf

buf.seek(0)
grep('filename') < buf

如果您想结合输入和输出重定向，由于Python喜欢处理比较链的方式，您必须将第一个重定向放在括号内

from pipepy import gzip
gzip = gzip(_text=False)
gzip < 'uncompressed.txt' > 'uncompressed.txt.gz'    # Wrong!
(gzip < 'uncompressed.txt') > 'uncompressed.txt.gz'  # Correct!

管道

《|》运算符用于自定义命令从哪里获取输入以及如何处理输出。根据操作数的类型，将出现不同的行为

4. 两个操作数都是《PipePy》实例

如果两个操作数都是命令，则结果将与在shell中发生的情况尽可能相似

from pipepy import git, grep
if git.diff(name_only=True) | grep('readme.txt'):
      print("readme was changed")

如果左操作数之前已评估，则将其输出（`stdout`）将直接作为输入传递给右操作数。否则，将并行执行两个命令，并将`left`的输出流式传输到`right`

5. 左操作数是任何类型的可迭代对象（包括字符串）

如果左操作数是任何类型的可迭代对象，则其元素将逐个馈送到命令的stdin

import random
from pipepy import grep

result = ["John is 18 years old\n", "Mary is 25 years old"] | grep("Mary")
print(result)
# <<< Mary is 25 years old

def my_stdin():
      for _ in range(500):
            yield f"{random.randint(1, 100)}\n"

result = my_stdin() | grep(17)
print(result)
# <<< 17
# ... 17
# ... 17
# ... 17
# ... 17

如果是字符串，它将一次性全部馈送

result = "John is 18 years old\nMary is 25 years old" | grep("Mary")

# Equivalent to

result = ["John is 18 years old\nMary is 25 years old"] | grep("Mary")

在这两种情况下，即在任何情况下，当右操作数是《PipePy》对象时，管道操作的返回值将是一个《未评估》的副本，该副本将在我们尝试访问其输出时进行评估。这意味着我们可以利用我们常用的后台功能

from pipepy import find, xargs
command = find('.') | xargs.wc
command = command.delay()

# Do something else in the meantime

for line in command:  # Here we wait for the command to finish
    linecount, wordcount, charcount, filename = line.split()
    # ...

这也意味着如果左操作数是可迭代的，则它将在命令评估时被消耗

from pipepy import grep

iterable = (line for line in ["foo\n", "bar\n"])
command = iterable | grep("foo")
command.stdout
# <<< 'foo\n'
list(iterable)
# <<< []

iterable = (line for line in ["foo\n", "bar\n"])
command = iterable | grep("foo")
list(iterable)  # Lets consume the iterable prematurely
# <<< ["foo\n", "bar\n"]
command.stdout
# <<< ''

此外，如果您更喜欢类似于函数调用的调用风格，即如果您想将命令的输入作为参数传递，您可以使用《_input》关键字参数

from pipepy import grep, ls

grep('setup', _input=ls)
# Is equivalent to
ls | grep('setup')

或使用方括号表示法

from pipepy import grep, ls

grep('setup')[ls]
# Is equivalent to
ls | grep('setup')

（我们使用圆括号作为参数，方括号作为输入，因为圆括号允许我们利用关键字参数，这对于命令行选项来说是一个很好的选择）

这适用于可迭代的输入和命令

7. 右操作数是函数

函数的参数需要是

《returncode》、《output》、《errors》的子集或
《stdout》、《stderr》的子集

由于函数的签名将被检查以分配正确的值，因此参数的顺序无关紧要

在这种情况下，将等待命令并使其评估后的输出可用于函数的参数

from pipepy import wc

def lines(output):
    for line in output.splitlines():
        try:
            lines, words, chars, filename = line.split()
        except ValueError:
            continue
        print(f"File {filename} has {lines} lines, {words} words and {chars} "
              "characters")

wc('*') | lines
# <<< File demo.py has 6 lines, 15 words and 159 characters
# ... File main.py has 174 lines, 532 words and 4761 characters
# ... File interactive2.py has 10 lines, 28 words and 275 characters
# ... File interactive.py has 12 lines, 34 words and 293 characters
# ... File total has 202 lines, 609 words and 5488 characters

在这种情况下，命令和函数将并行执行，并将命令的`stdout`和`stderr`流提供给函数

import re
from pipepy import ping

def mean_ping(stdout):
    pings = []
    for line in stdout:
        match = re.search(r'time=([\d\.]+) ms$', line.strip())
        if not match:
            continue
        time = float(match.groups()[0])
        pings.append(time)
        if len(pings) % 10 == 0:
            print(f"Mean time is {sum(pings) / len(pings)} ms")

ping('-c', 30, "google.com") | mean_ping
# >>> Mean time is 71.96000000000001 ms
# ... Mean time is 72.285 ms
# ... Mean time is 72.19666666666667 ms

如果命令在函数之前结束，则`next(stdout)`将引发《StopIteration》。如果函数在命令之前结束，则命令的`stdin`将被关闭

管道操作的返回值将是函数的返回值。函数甚至可以包含《yield》一词，从而返回一个可以管道到另一个命令的生成器

将这些放在一起，我们可以做如下事情

from pipepy import cat, grep

def my_input():
    yield "line one\n"
    yield "line two\n"
    yield "line two\n"
    yield "something else\n"
    yield "line three\n"

def my_output(stdout):
    for line in stdout:
        yield line.upper()

print(my_input() | cat | grep('line') | my_output | grep("TWO"))
# <<< LINE TWO
# ... LINE TWO

8. 右操作数是生成器

这是管道的一种更奇特的形式。在这里，我们利用Python的《将值传递给生成器》功能。原始生成器必须使用《a = (yield b)》语法发送和接收数据。管道操作的结果将是一个新的生成器，它将产生原始生成器产生的任何内容，同时，在原始生成器中，每个《yield》命令的返回值将是《PipePy》实例的下一行非空内容

from pipepy import echo

def upperize():
    line = yield
    while True:
        line = (yield line.upper())

# Remember, `upperize` is a function, `upperize()` is a generator
list(echo("aaa\nbbb") | upperize())
# <<< ["AAA\n", "BBB\n"]

由于管道操作的返回值是生成器，因此它可以管道到另一个命令

print(echo("aaa\nbbb") | upperize() | grep("AAA"))
# <<< AAA

与后台进程交互

与后台进程交互有3种方式：仅《读取》、《写入》和《读写》。我们已经讨论了仅《读取》和仅《写入》

1. 逐步向命令发送数据

这通过将可迭代对象传递到命令行来实现。实际上，该命令与可迭代对象并行运行，并且随着数据的可用，将数据馈送到命令中。我们将对之前的示例进行轻微修改，以更好地演示这一点。

import random
import time
from pipepy import grep

def my_stdin():
    start = time.time()
    for _ in range(500):
        time.sleep(.01)
        yield f"{time.time() - start} {random.randint(1, 100)}\n"

command = my_stdin() | grep('-E', r'\b17$', _stream_stdout=True)
command()
# <<< 0.3154888153076172 17
# ... 1.5810892581939697 17
# ... 1.7773401737213135 17
# ... 2.8303775787353516 17
# ... 3.4419643878936768 17
# ... 4.511774301528931  17

在这里，grep 实际上是与生成器并行运行的，并且随着匹配项的出现而打印出来，因为命令的输出是通过 _stream_stdout 参数流式传输到控制台的（关于这一点，请参阅下面的说明）。

2. 从命令中增量读取数据

这可以通过将命令的输出传递到具有其子集 stdin、stdout 和 stderr 作为其参数的函数，或者像我们之前演示的那样，通过迭代命令的输出

import time
from pipepy import ping

start = time.time()
for line in ping('-c', 3, 'google.com'):
    print(time.time() - start, line.strip().upper())
# <<< 0.15728354454040527 PING GOOGLE.COM (172.217.169.142) 56(84) BYTES OF DATA.
# ... 0.1574106216430664  64 BYTES FROM SOF02S32-IN-F14.1E100.NET (172.217.169.142): ICMP_SEQ=1 TTL=103 TIME=71.8 MS
# ... 1.1319730281829834  64 BYTES FROM 142.169.217.172.IN-ADDR.ARPA (172.217.169.142): ICMP_SEQ=2 TTL=103 TIME=75.3 MS
# ... 2.1297826766967773  64 BYTES FROM 142.169.217.172.IN-ADDR.ARPA (172.217.169.142): ICMP_SEQ=3 TTL=103 TIME=73.4 MS
# ... 2.129857063293457
# ... 2.129875659942627   --- GOOGLE.COM PING STATISTICS ---
# ... 2.1298911571502686  3 PACKETS TRANSMITTED, 3 RECEIVED, 0% PACKET LOSS, TIME 2004MS
# ... 2.129910707473755   RTT MIN/AVG/MAX/MDEV = 71.827/73.507/75.253/1.399 MS

同样，ping 命令实际上是与 for 循环的主体并行运行的，并且随着每一行的出现将其提供给 for 循环的主体。

3. 从命令读取数据并写入数据

假设我们有一个命令，让用户参加数学测验。与该命令的正常交互可能如下所示

→ math_quiz
3 + 4 ?
→ 7
Correct!
8 + 2 ?
→ 12
Wrong!
→ Ctrl-d

可以使用 with 语句使用 Python 与该命令以读写方式交互

from pipepy import math_quiz

result = []
with math_quiz as (stdin, stdout, stderr):
    stdout = (line.strip() for line in stdout if line.strip())
    try:
        for _ in range(3)
            question = next(stdout)
            a, _, b, _ = question.split()
            answer = str(int(a) + int(b))
            stdin.write(answer + "\n")
            stdin.flush()
            verdict = next(stdout)
            result.append((question, answer, verdict))
    except StopIteration:
        pass

result
# <<< [('10 + 7 ?', '17', 'Correct!'),
# ...  ('5 + 5 ?', '10', 'Correct!'),
# ...  ('5 + 5 ?', '10', 'Correct!')]

stdin、stdout 和 stderr 是后台进程的打开文件流。当 with 块的主体结束时，会向进程发送 EOF 并等待它。

需要记住，如果命令期望它，请将 stdin 中馈送到行末尾的换行符。此外，别忘了不时调用 stdin.flush()。

可以在涉及 PipePy 对象的管道表达式中调用 with。在这种情况下，每个 PipePy 对象的 stdout 将连接到下一个对象的 stdin，提供给 with 块主体的 stdin 将是左侧命令的 stdin，提供给 with 块主体的 stdout 和 stderr 将是右侧命令的 stdout 和 stderr。

from pipepy import cat, grep

command = cat | grep("foo") | cat | cat | cat  # We might as well keep going
with command as (stdin, stdout, stderr):
    stdin.write("foo1\n")
    stdin.write("bar2\n")
    stdin.write("foo3\n")
    stdin.close()
    assert next(stdout).strip() == "foo1"
    assert next(stdout).strip() == "foo3"

修改命令行为

二进制模式

所有命令都在文本模式下执行，这意味着它们处理 str 对象。这可能会引起问题。例如

from pipepy import gzip
result = "hello world" | gzip
print(result.stdout)
# <<< Traceback (most recent call last):
# ... ...
# ... UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8b in position 1: invalid start byte

gzip 不能在文本模式下工作，因为其输出是二进制数据，无法进行 utf-8 解码。当不希望使用文本模式时，可以将命令转换为二进制模式，将其 _text 参数设置为 False

from pipepy import gzip
gzip = gzip(_text=False)
result = "hello world" | gzip
print(result.stdout)
# <<< b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xe1\x02\x00-;\x08\xaf\x0c\x00\x00\x00'

输入和输出将通过使用 'UTF-8' 编码转换为二进制。在上面的示例中，我们的输入类型是 str，在馈送到 gzip 之前将其 utf-8 编码。您可以使用 _encoding 关键字参数更改编码。

from pipepy import gzip
gzip = gzip(_text=False)
result = "καλημέρα" | gzip
print(result.stdout)
# <<< b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03\x01\x10\x00\xef\xff\xce\xba\xce\xb1\xce\xbb\xce\xb7\xce\xbc\xce\xad\xcf\x81\xce\xb1"\x15g\xab\x10\x00\x00\x00'
result = "καλημέρα" | gzip(_encoding="iso-8859-7")
print(result.stdout)
# <<< b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03{\xf5\xf0\xf5\xf37w?>\x04\x00\x1c\xe1\xc0\xf7\x08\x00\x00\x00'

流式传输到控制台

在调用时，可以将 _stream_stdout 和 _stream_stderr 关键字参数设置为 True。这意味着相应的流不会被结果捕获，而是流式传输到控制台。这允许用户与交互式命令交互。考虑以下两个示例

fzf 的工作方式如下
1. 它从其 stdin 收集选择列表
2. 它在 stderr 上显示选择项，并根据用户输入不断刷新它
3. 它直接开始捕获键盘上的按键，绕过 stdin，以允许用户做出选择。
4. 当用户按下 Enter 时，它将选择项打印到其 stdout
考虑到所有这些，我们可以这样做
```
from pipepy import fzf
fzf = fzf(_stream_stderr=True)

# This will open an fzf session to let us choose between "John" and "Mary"
print("John\nMary" | fzf)
# <<< Mary
```

dialog 与 fzf 的工作方式类似，但将 stdout 与 stderr 交换

它从其参数中收集选择列表
它在 stdout 上显示选择，并根据用户输入不断刷新
它直接开始捕获键盘上的按键，绕过 stdin，以允许用户做出选择。
当用户按下 Enter 时，它将选择打印到其 stderr

考虑到所有这些，我们可以这样做

from pipepy import dialog
dialog = dialog(_stream_stdout=True)

# This will open a dialog session to let us choose between "John" and "Mary"
result = dialog(checklist=True)('Choose name', 30, 110, 0,
                                "John", '', "on",
                                "Mary", '', "off")
print(result.stderr)
# <<< John

此外，在脚本中，您可能不感兴趣捕获命令的输出，但可能希望将其流式传输到控制台以向用户显示命令的输出。您可以通过设置 _stream 参数强制命令流式传输其整个输出

from pipepy import wget

wget('https://...', _stream=True)()

虽然 stdout 和 stderr 不会被捕获，但 returncode 会，因此您仍然可以在布尔表达式中使用该命令

from pipepy import wget

if wget('https://...', _stream=True):
     print("Download succeeded")
else:
     print("Download failed")

您可以通过调用 pipepy.set_always_stream(True) 来使将输出流式传输到控制台成为默认行为。在某些情况下这可能是有用的，例如 Makefiles（见下文）。

import pipepy
from pipepy import ls
pipepy.set_always_stream(True)
ls()  # Alsost equivalent to `ls(_stream=True)()`
pipepy.set_always_stream(False)

与设置 _stream=True 强制命令将输出流式传输到控制台类似，设置 _stream=False 强制即使已调用 set_always_stream 也捕获其输出

import pipepy
from pipepy import ls

pipepy.set_always_stream(True)
ls()                 # Will stream its output
ls(_stream=False)()  # Will capture its output
pipepy.set_always_stream(False)

异常

您可以在已评估的结果上调用 .raise_for_returncode() 来在返回码不是 0 时引发异常（想想 requests 的 .raise_for_status()）

from pipepy import ping, PipePyError
result = ping("asdf")()  # Remember, we have to evaluate it first

result.raise_for_returncode()
# <<< PipePyError: (2, '', 'ping: asdf: Name or service not known\n')

try:
    result.raise_for_returncode()
except PipePyError as exc:
    print(exc.returncode)
    # <<< 2
    print(exc.stdout)
    # <<< ""
    print(exc.stderr)
    # <<< ping: asdf: Name or service not known

您可以通过调用 pipepy.set_always_raise(True) 使所有命令在返回码不是零时引发异常。

import pipepy
from pipepy import ping
pipepy.set_always_raise(True)
ping("asdf")()
# <<< PipePyError: (2, '', 'ping: asdf: Name or service not known\n')

如果已设置“始终引发”，您仍然可以通过设置 _raise=False 强制命令抑制其异常

import pipepy
from pipepy import ping
pipepy.set_always_raise(True)
try:
    ping("asdf")()  # Will raise an exception
except Exception as exc:
    print(exc)
# <<< PipePyError: (2, '', 'ping: asdf: Name or service not known\n')

try:
    ping("asdf", _raise=False)()  # Will not raise an exception
except Exception as exc:
    print(exc)

“交互”模式

当设置“交互”模式时，__repr__ 方法将简单地返回 self.stdout + self.stderr。这为交互式 Python shell 提供了一些基本功能。要设置交互模式，请运行 pipepy.set_interactive(True)

import pipepy
from pipepy import ls, overload_chars
pipepy.set_interactive(True)
ls
# <<< demo.py
# ... interactive2.py
# ... interactive.py
# ... main.py

overload_chars(locals())
ls -l
# <<< total 20
# ... -rw-r--r-- 1 kbairak kbairak  159 Feb  7 22:05 demo.py
# ... -rw-r--r-- 1 kbairak kbairak  275 Feb  7 22:04 interactive2.py
# ... -rw-r--r-- 1 kbairak kbairak  293 Feb  7 22:04 interactive.py
# ... -rw-r--r-- 1 kbairak kbairak 4761 Feb  8 20:42 main.py

永久更改

由于 PipePy 对象将其参数列表视为简单地传递给 subprocess.Popen 函数的字符串列表，并且尽管技术上它是正在执行的命令，但第一个参数没有特殊意义，因此您可以创建包含我们讨论的更改的 PipePy 实例，并将它们用作将继承这些更改的命令的模板

stream_sh = PipePy(_stream=True)
stream_sh
# <<< PipePy()
stream_sh._stream
# <<< True

stream_sh.ls
# <<< PipePy('ls')
stream_sh.ls._stream
# <<< True

r = stream_sh.ls()
# <<< check_tag.py  Makefile.py     setup.cfg  tags
# ... htmlcov       pyproject.toml  setup.py   test_requirements.txt
# ... LICENSE       README.md       src

r.stdout
# <<< None

r.returncode
# <<< 0

raise_sh = PipePy(_raise=True)
raise_sh
# <<< PipePy()
raise_sh.false
# <<< PipePy('false')
raise_sh.false()
# <<< Traceback (most recent call last):
# ... ...
# ... pipepy.exceptions.PipePyError: (1, '', '')

这可以作为 set_always_stream 和 set_always_raise 的更封闭的替代方案。

杂项

.terminate()、.kill() 和 .send_signal() 简单地转发方法调用到底层 Popen 对象。

以下是 pipepy 内部实现的一些实用程序，它们不使用 shell 子进程，但我们认为它们对脚本编写很有用。

`cd`

在其最简单形式中，pipepy.cd 是 os.chdir 的别名

from pipepy import cd, pwd

print(pwd())
# <<< /foo

cd('bar')
print(pwd())
# <<< /foo/bar

cd('..')
print(pwd())
# <<< /foo

但它也可以用作临时目录更改的上下文处理器

print(pwd())
# <<< /foo

with cd("bar"):
    print(pwd())
# <<< /foo/bar

print(pwd())
# <<< /foo

`export`

在其最简单形式中，pipepy.export 是 os.environ.update 的别名

import os
from pipepy import export

print(os.environ['HOME'])
# <<< /home/foo

export(PATH="/home/foo/bar")
print(os.environ['HOME'])
# <<< /home/foo/bar

但它也可以用作临时环境更改的上下文处理器

print(os.environ['HOME'])
# <<< /home/foo

with export(PATH="/home/foo/bar"):
    print(os.environ['HOME'])
# <<< /home/foo/bar

print(os.environ['HOME'])
# <<< /home/foo

如果 with 块体内进一步修改了环境变量，则退出时不会将其还原

with export(PATH="/home/foo/bar"):
    export(PATH="/home/foo/BAR")

print(os.environ['HOME'])
# <<< /home/foo/BAR

`source`

source 函数运行 bash 脚本，提取在脚本中设置的并保存在当前环境中的结果环境变量。与 export 类似，它可以用作上下文处理器（实际上，它使用 export 内部）

# env
export AAA=aaa

import os
from pipepy import source

with source('env'):
    print(os.environ['AAA'])
# <<< aaa
'AAA' in os.environ
# <<< False

source('env')
print(os.environ['AAA'])
# <<< aaa

source 函数有以下关键字参数可用

递归 (布尔值，默认为 False): 如果设置为 true，则当前目录及其所有父目录中同名文件将被源文件，按反向顺序。这允许环境变量的嵌套。

- /
  |
  + - home/
      |
      - kbairak/
        |
        + - env:
        |     export COMPOSE_PROJECT_NAME="pipepy"
        |
        + - project/
            |
            + - env:
                  export COMPOSE_FILE="docker-compose.yml:docker-compose-dev.yml"

from pipepy import cd, source, docker_compose
cd('/home/kbairak/project')
source('env', recursive=True)
# Now I have both `COMPOSE_PROJECT_NAME` and `COMPOSE_FILE`

按照顺序源文件 /home/kbairak/env 和 /home/kbairak/project/env。

安静 (布尔值，默认为 True): 如果源文件失败，source 通常会跳过其源文件而不报错，并继续下一个（如果设置了 recursive）。使用 quiet=False 时，将引发异常且环境不会更新。
shell (字符串，默认为 'bash')：用于执行源文件的 shell 命令。

pymake

此库附带一个名为 pymake 的命令，旨在尽可能复制 GNU make 的语法和行为，但使用 Python 实现。一个 Makefile.py 文件看起来像这样（这实际上是当前库 Makefile 的一部分）

import pipepy
from pipepy import python, rm

pipepy.set_always_stream(True)
pipepy.set_always_raise(True)

def clean():
    rm('-rf', "build", "dist")()

def build(clean):
    python('-m', "build")()

def publish(build):
    python('-m', "twine").upload("dist/*")()

现在可以运行 pymake publish 来运行 publish make 目标及其依赖项。函数参数的名称用于定义依赖项，因此 clean 是 build 的依赖项，而 build 是 publish 的依赖项。

（您不必在 Makefile.py 中使用 pipepy 命令，但不得不说这是一个非常好的匹配）

参数包含依赖目标返回的任何值

def a():
    return 1

def b():
    return 2

def c(a, b):
    print(a + b)

→ pymake c
# ← 3

每个依赖项最多执行一次，即使它被用作多次依赖项

def a():
    print("pymake target a")

def b(a):
    print("pymake target b")

def c(a, b):
    print("pymake target c")

→ pymake c
# ← pymake target a
# ← pymake target b
# ← pymake target c

您可以将全局变量 DEFAULT_PYMAKE_TARGET 设置为定义默认目标。

from pipepy import pytest

DEFAULT_PYMAKE_TARGET = "test"

def test():
    pytest(_stream=True)()

`pymake` 变量

除了依赖项外，您还可以使用函数参数定义可以由 pymake 调用覆盖的变量。这可以通过两种方式完成

使用函数的键值参数

# Makefile.py

def greeting(msg="world"):
    print(f"hello {msg}")

→ pymake greeting
# ← hello world

→ pymake greeting msg=Bill
# ← hello Bill

使用在 Makefile.py 中定义的全局变量

# Makefile.py

msg = "world"

def greeting():
    print(f"hello {msg}")

→ pymake greeting
# ← hello world

→ pymake greeting msg=Bill
# ← hello Bill

`pymake` 的 shell 完成功能

pymake 支持 bash 和 zsh 的 shell 完成功能。

在 bash 中运行

eval $(pymake --setup-bash-completion)

然后您将能够看到如下内容（以下是从 pipepy 的 Makefile 中摘取的示例）

[kbairak@kbairakdelllaptop pipepy]$ pymake <TAB><TAB>
build      clean      debugtest  publish    watchtest
checks     covtest    html       test

在 zsh 中运行

eval $(pymake --setup-zsh-completion)

然后您将能够看到如下内容（以下是从 pipepy 的 Makefile 中摘取的示例）

(pipepy) ➜  pipepy git:(master) ✗ pymake <TAB>
build      -- Build package
checks     -- Run static checks on the code (flake8, isort)
clean      -- Clean up build directories
covtest    -- Run tests and produce coverge report
debugtest  -- Run tests without capturing their output. This makes using an interactive debugger possible
html       -- Run tests and open coverage report in browser
publish    -- Publish package to PyPI
test       -- Run tests
watchtest  -- Automatically run tests when a source file changes

描述来自 pymake 目标的文档字符串。

您可以将 eval 语句放入您的 .bashrc/.zshrc 中。

待办事项

等待的超时
从/到文件-like 对象重定向输入/输出
同时流式传输和捕获（文件-like 对象的包装类）

with 块，其中 PipePy 调用将转发到上下文的 stdin，例如

from pipepy import ssh
with ssh("some-host") as host:
    r = host.ls()  # Will actually send 'ls\n' to ssh's stdin

项目详情

这些详情尚未由PyPI 验证

项目链接

主页

发布历史发布通知 | RSS 源

此版本

0.0.11

2022 年 9 月 20 日

0.0.10

2021 年 8 月 16 日

0.0.9

2021 年 3 月 14 日

0.0.8

2021 年 2 月 27 日

0.0.7

2021 年 2 月 16 日

0.0.6

2021 年 2 月 11 日

0.0.5

2021 年 2 月 10 日

0.0.1

2021 年 2 月 9 日

下载文件

下载您平台上的文件。如果您不确定要选择哪个，请了解更多关于安装软件包的信息。

源分发

pipepy-0.0.11.tar.gz (65.2 kB 查看哈希值)

上传时间 2022 年 9 月 20 日 源

构建分发

pipepy-0.0.11-py3-none-any.whl (45.0 kB 查看哈希值)

上传于 2022年9月20日 Python 3

散列信息 for pipepy-0.0.11.tar.gz

散列信息 for pipepy-0.0.11.tar.gz
算法	散列摘要
SHA256	`34fc109f21db01b7c47cbb2cc752b62a514b93b209f3979186ea21c7ce890cf9`
MD5	`17e585cef21a3134fc5e26bfaa4b8edd`
BLAKE2b-256	`ad2644fed51cd16e9fd56a028a51c3ee84d27a2dc88998e58a759a77b6beecb0`

散列信息 for pipepy-0.0.11-py3-none-any.whl

散列信息 for pipepy-0.0.11-py3-none-any.whl
算法	散列摘要
SHA256	`0ea7f1642103c10d549f97dd6226d43fb9b44f396920c8cbf330017d14ed0e4a`
MD5	`17ed153ec2c7e60a90c844f15f5fd45b`
BLAKE2b-256	`dbda5b15748bacea1c3a92a4a484d7f57a956f54b4b61eb1c995db3de7e13eda`

pipepy 0.0.11

导航

验证详情

维护者

未经验证的详情

项目链接

元数据

分类器

项目描述

目录

为什么？与其他类似框架的比较

安装和测试

简介，基本用法

自定义命令

懒惰

后台命令

重定向文件输出

管道

4. 两个操作数都是《PipePy》实例

5. 左操作数是任何类型的可迭代对象（包括字符串）

7. 右操作数是函数

8. 右操作数是生成器

与后台进程交互

1. 逐步向命令发送数据

2. 从命令中增量读取数据

3. 从命令读取数据并写入数据

修改命令行为

二进制模式

流式传输到控制台

异常

“交互”模式

永久更改

杂项

cd

export

source

pymake

pymake 变量

pymake 的 shell 完成功能

待办事项

项目详情

验证详情

维护者

未经验证的详情

项目链接

元数据

分类器

发布历史 发布通知 | RSS 源

下载文件

源分发

构建分发

`cd`

`export`

`source`

`pymake` 变量

`pymake` 的 shell 完成功能

发布历史发布通知 | RSS 源