| Name | recmd JSON |
| Version |
0.0.3
JSON |
| download |
| home_page | None |
| Summary | Inspired by zx command constructor |
| upload_time | 2024-06-28 15:41:29 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.12 |
| license | MIT License Copyright (c) 2024 rewirepy Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
command
shell
subprocess
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 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
```py
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
```py
import sys
from recmd.shell import sh
@sh
def python(code: str, *args):
return sh(f"{sys.executable} -c {code} {args:*}")
```
#### Running commands
Sync:
```py
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:
```py
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:
```py
# 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:
```py
# 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:
```py
# 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:
```py
# 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"
```
Raw data
{
"_id": null,
"home_page": null,
"name": "recmd",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "command, shell, subprocess",
"author": null,
"author_email": "Ivan Vozhakov <gou177@bk.ru>",
"download_url": "https://files.pythonhosted.org/packages/fb/10/f80a04da41a4ae5ff13c4be3b54cb6c9b53549a409bbe2463e24ac27291a/recmd-0.0.3.tar.gz",
"platform": null,
"description": "# Recmd: Inspired by zx command constructor\r\n\r\nEasy to use, type-safe (pyright), sync and async, and cross-platform command executor\r\n\r\n## Installation\r\n\r\nInstall recmd using pip: `pip install recmd`\r\nWith async support: `pip install recmd[async]`\r\n\r\n## Usage\r\n\r\n#### Convert formatted strings to argument lists\r\n\r\n```py\r\nfrom recmd import shell, sh\r\n\r\n@sh\r\ndef argument_list(value: str, *args):\r\n return shell(f\"executable arguments --value {value} 'more arguments with {value}' {args:*!s}\")\r\n # :*!s converts args into `*[f\"{x!s}\" for x in args]`\r\n # if !s is omitted then it turned into just `*args`\r\n # you can also add any python format after :* (:*:.2f)\r\n \r\nassert argument_list(\"test asd\", 1, 2, 3) == [\r\n \"executable\", \"arguments\", \"--value\", \"test asd\", \"more arguments with test asd\", \"1\", \"2\", \"3\"\r\n]\r\n```\r\n\r\n#### Constructing commands\r\n\r\n```py\r\nimport sys\r\nfrom recmd.shell import sh\r\n\r\n@sh\r\ndef python(code: str, *args):\r\n return sh(f\"{sys.executable} -c {code} {args:*}\")\r\n\r\n```\r\n\r\n#### Running commands\r\n\r\nSync:\r\n\r\n```py\r\nfrom recmd.executor.subprocess import SubprocessExecutor\r\n\r\n# set globally (context api)\r\nSubprocessExecutor.context.set(SubprocessExecutor())\r\n\r\n# set for code block\r\nwith SubprocessExecutor().use():\r\n ...\r\n\r\n\r\n# `~` runs and waits for process to exit, then `assert` checks that exit code == 0\r\nassert ~python(\"pass\")\r\n\r\n# you can also use process as context manager\r\nwith python(\"pass\"):\r\n pass\r\n\r\n```\r\n\r\nAsync:\r\n\r\n```py\r\nimport anyio\r\nfrom recmd.executor.anyio import AnyioExecutor\r\n\r\n# set globally (context api)\r\nAnyioExecutor.context.set(AnyioExecutor())\r\n\r\n# set for code block\r\nwith AnyioExecutor().use():\r\n ...\r\n\r\nasync def run():\r\n # `await` runs and waits for process to exit, then `assert` checks that exit code == 0\r\n assert await python(\"pass\")\r\n\r\n # you can also use process as context manager\r\n async with python(\"pass\"):\r\n pass\r\n\r\nanyio.run(run)\r\n```\r\n\r\n#### Interacting with processes\r\n\r\nSync:\r\n\r\n```py\r\n# send to stdin and read from stdout\r\nassert ~python(\"print(input(),end='')\").send(\"123\").output() == \"123\"\r\n\r\nfrom recmd import IOStream\r\n\r\n\r\n# manually control streams\r\nwith IOStream() >> python(\"print(input(),end='')\") >> IOStream() as process:\r\n process.stdin.sync_io.write(b\"hello\")\r\n process.stdin.sync_io.close()\r\n assert process.stdout.sync_io.read() == b\"hello\"\r\n```\r\n\r\nAsync:\r\n\r\n```py\r\n# send to stdin and read from stdout\r\nassert await python(\"print(input(),end='')\").send(\"123\").output() == \"123\"\r\n\r\nfrom recmd import IOStream\r\n\r\n\r\n# manually control streams\r\nasync with IOStream() >> python(\"print(input(),end='')\") >> IOStream() as process:\r\n await process.stdin.async_write.send(b\"hello\")\r\n await process.stdin.async_write.aclose()\r\n assert await process.stdout.async_read.receive() == \"bhello\"\r\n```\r\n\r\n#### Pipes\r\n\r\nSync:\r\n\r\n```py\r\n# redirect stdout from first process to stdin of second\r\nfrom recmd import Capture\r\n\r\ngroup = ~(python(\"print(123)\") | python(\"print(input(),end='')\") >> Capture())\r\n\r\nwith python(\"print(123)\") | python(\"print(input(),end='')\") >> Capture() as group:\r\n ...\r\n\r\nassert group.commands[-1].stdout.get() == b\"123\"\r\n```\r\n\r\nAsync:\r\n\r\n```py\r\n# redirect stdout from first process to stdin of second\r\nfrom recmd import Capture\r\n\r\ngroup = await (python(\"print(123)\") | python(\"print(input(),end='')\") >> Capture())\r\n\r\nasync with python(\"print(123)\") | python(\"print(input(),end='')\") >> Capture() as group:\r\n ...\r\n\r\nassert group.commands[-1].stdout.get() == b\"123\"\r\n```\r\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2024 rewirepy Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": "Inspired by zx command constructor",
"version": "0.0.3",
"project_urls": {
"Homepage": "https://github.com/gou177/recmd"
},
"split_keywords": [
"command",
" shell",
" subprocess"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4fb8747b519b8fc47ef43c9840ac6fd84d893534fea9eb8d37abc1d32c162dc0",
"md5": "d10ce4ed24c1cf1201e20544683e62e9",
"sha256": "84f0bda358ad9ea2c8ce3ad65b64a37e6701eedef8d5d2012dda2ab9d48205b9"
},
"downloads": -1,
"filename": "recmd-0.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d10ce4ed24c1cf1201e20544683e62e9",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 16254,
"upload_time": "2024-06-28T15:41:28",
"upload_time_iso_8601": "2024-06-28T15:41:28.114901Z",
"url": "https://files.pythonhosted.org/packages/4f/b8/747b519b8fc47ef43c9840ac6fd84d893534fea9eb8d37abc1d32c162dc0/recmd-0.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "fb10f80a04da41a4ae5ff13c4be3b54cb6c9b53549a409bbe2463e24ac27291a",
"md5": "29d34c0b1b0f8db9e78da301d2b7c327",
"sha256": "24aa347ccfb7bb49f0f5d18591afdb6d4023b6788663fa4fcc5b292611a52353"
},
"downloads": -1,
"filename": "recmd-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "29d34c0b1b0f8db9e78da301d2b7c327",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 16461,
"upload_time": "2024-06-28T15:41:29",
"upload_time_iso_8601": "2024-06-28T15:41:29.484937Z",
"url": "https://files.pythonhosted.org/packages/fb/10/f80a04da41a4ae5ff13c4be3b54cb6c9b53549a409bbe2463e24ac27291a/recmd-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-28 15:41:29",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "gou177",
"github_project": "recmd",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "recmd"
}
