Inspired by zx command constructor
Project description
Recmd: Inspired by zx command constructor
Easy to use, type-safe (pyright), sync and async, and cross-platform command executor
Installation
Install recmd using pip: pip install recmd
With async support: pip install recmd[async]
Usage
Convert formatted strings to argument lists
from recmd import shell, sh
@sh
def argument_list(value: str, *args):
return shell(f"executable arguments --value {value} 'more arguments with {value}' {args:*!s}")
# :*!s converts args into `*[f"{x!s}" for x in args]`
# if !s is omitted then it turned into just `*args`
# you can also add any python format after :* (:*:.2f)
assert argument_list("test asd", 1, 2, 3) == [
"executable", "arguments", "--value", "test asd", "more arguments with test asd", "1", "2", "3"
]
Constructing commands
import sys
from recmd.shell import sh
@sh
def python(code: str, *args):
return sh(f"{sys.executable} -c {code} {args:*}")
Running commands
Sync:
from recmd.executor.subprocess import SubprocessExecutor
# set globally (context api)
SubprocessExecutor.context.set(SubprocessExecutor())
# set for code block
with SubprocessExecutor().use():
...
# `~` runs and waits for process to exit, then `assert` checks that exit code == 0
assert ~python("pass")
# you can also use process as context manager
with python("pass"):
pass
Async:
import anyio
from recmd.executor.anyio import AnyioExecutor
# set globally (context api)
AnyioExecutor.context.set(AnyioExecutor())
# set for code block
with AnyioExecutor().use():
...
async def run():
# `await` runs and waits for process to exit, then `assert` checks that exit code == 0
assert await python("pass")
# you can also use process as context manager
async with python("pass"):
pass
anyio.run(run)
Interacting with processes
Sync:
# send to stdin and read from stdout
assert ~python("print(input(),end='')").send("123").output() == "123"
from recmd import IOStream
# manually control streams
with IOStream() >> python("print(input(),end='')") >> IOStream() as process:
process.stdin.sync_io.write(b"hello")
process.stdin.sync_io.close()
assert process.stdout.sync_io.read() == b"hello"
Async:
# send to stdin and read from stdout
assert await python("print(input(),end='')").send("123").output() == "123"
from recmd import IOStream
# manually control streams
async with IOStream() >> python("print(input(),end='')") >> IOStream() as process:
await process.stdin.async_write.send(b"hello")
await process.stdin.async_write.aclose()
assert await process.stdout.async_read.receive() == "bhello"
Pipes
Sync:
# redirect stdout from first process to stdin of second
from recmd import Capture
group = ~(python("print(123)") | python("print(input(),end='')") >> Capture())
with python("print(123)") | python("print(input(),end='')") >> Capture() as group:
...
assert group.commands[-1].stdout.get() == b"123"
Async:
# redirect stdout from first process to stdin of second
from recmd import Capture
group = await (python("print(123)") | python("print(input(),end='')") >> Capture())
async with python("print(123)") | python("print(input(),end='')") >> Capture() as group:
...
assert group.commands[-1].stdout.get() == b"123"
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
recmd-0.0.4.tar.gz
(16.6 kB
view hashes)
Built Distribution
recmd-0.0.4-py3-none-any.whl
(16.3 kB
view hashes)