# Finally, a subprocess type that streams out stdout/stderr easily
[![Win_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_win.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_win.yml)
[![Ubuntu_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_ubuntu.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_ubuntu.yml)
[![MacOS_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_macos.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_macos.yml)
Capturing the stderr AND stdout from a process in python is not that easy.
This class makes this capturing much easier by delegating the line capturing
to seperate threads. This capture can be totally in memory or can optionally
be streamed to a output stream such as a file handle.
This class will unconditionally launch a shell command and the input will always
be string, not an array like what is accepted by subprocess.Popen().
# Example:
Super simple example:
```python
from capturing_process import CapturingProcess
out_stream = StringIO()
p = CapturingProcess("echo hi", stdout=out_stream)
p.wait()
self.assertIn("hi", out_stream.getvalue())
self.assertIn("hi", p.get_stdout())
```
For splitting the output to stdout and a file you'd write a stream class like so:
```python
import logging
class MyStream:
def __init__(self) -> None:
pass
def write(self, data: str) -> None:
logging.info(data.rstrip('\n'))
print(data, end="")
out_stream = MyStream()
proc = CapturingProcess("echo hi", stdout=out_stream)
proc.wait() # Output will be captured in logging file and stdout
```
To silence an output stream (stdout/stderr) drop a StringIO object as an argument to
the CapturingProcess like so:
## If you want the entire stdout/stderr bytes
```python
proc.get_stdout()
proc.get_stderr()
```
# Python version: 3.6+
Because of the use of type annotations, this library is not compatible with python 2.7
However you are free to strip out these type annotations and this project *should* work
pretty well.
# Links:
* https://pypi.org/project/capturing-process/
* https://github.com/zackees/capturing_process
# Versions
* 1.0.9: stdout/stderr threads are now forcefully killed within .1 second if they don't join.
* 1.0.8: Fixes CapturingProcess.kill blocking if the stdout and stderr threads fail to join.
Raw data
{
"_id": null,
"home_page": "https://github.com/zackees/capturing_process",
"name": "capturing-process",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.6.0",
"maintainer_email": null,
"keywords": null,
"author": "Zach Vorhies",
"author_email": "dont@email.me",
"download_url": "https://files.pythonhosted.org/packages/68/0c/ee2ddde6d9e4de7752ce899ee7d9aa836834161086ce47fefb793603a341/capturing_process-1.0.13.tar.gz",
"platform": null,
"description": "# Finally, a subprocess type that streams out stdout/stderr easily\r\n\r\n[![Win_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_win.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_win.yml)\r\n[![Ubuntu_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_ubuntu.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_ubuntu.yml)\r\n[![MacOS_Tests](https://github.com/zackees/capturing_process/actions/workflows/push_macos.yml/badge.svg)](https://github.com/zackees/capturing_process/actions/workflows/push_macos.yml)\r\n\r\nCapturing the stderr AND stdout from a process in python is not that easy.\r\nThis class makes this capturing much easier by delegating the line capturing\r\nto seperate threads. This capture can be totally in memory or can optionally\r\nbe streamed to a output stream such as a file handle.\r\n\r\nThis class will unconditionally launch a shell command and the input will always\r\nbe string, not an array like what is accepted by subprocess.Popen().\r\n\r\n# Example:\r\n\r\nSuper simple example:\r\n\r\n```python\r\nfrom capturing_process import CapturingProcess\r\n\r\nout_stream = StringIO()\r\np = CapturingProcess(\"echo hi\", stdout=out_stream)\r\np.wait()\r\nself.assertIn(\"hi\", out_stream.getvalue())\r\nself.assertIn(\"hi\", p.get_stdout())\r\n```\r\n\r\nFor splitting the output to stdout and a file you'd write a stream class like so:\r\n\r\n```python\r\nimport logging\r\nclass MyStream:\r\n def __init__(self) -> None:\r\n pass\r\n\r\n def write(self, data: str) -> None:\r\n logging.info(data.rstrip('\\n'))\r\n print(data, end=\"\")\r\n\r\n\r\nout_stream = MyStream()\r\nproc = CapturingProcess(\"echo hi\", stdout=out_stream)\r\nproc.wait() # Output will be captured in logging file and stdout\r\n```\r\n\r\n\r\nTo silence an output stream (stdout/stderr) drop a StringIO object as an argument to\r\nthe CapturingProcess like so:\r\n\r\n\r\n## If you want the entire stdout/stderr bytes\r\n\r\n```python\r\nproc.get_stdout()\r\nproc.get_stderr()\r\n```\r\n\r\n# Python version: 3.6+\r\n\r\nBecause of the use of type annotations, this library is not compatible with python 2.7\r\nHowever you are free to strip out these type annotations and this project *should* work\r\npretty well.\r\n\r\n# Links:\r\n\r\n * https://pypi.org/project/capturing-process/\r\n * https://github.com/zackees/capturing_process\r\n\r\n\r\n# Versions\r\n\r\n * 1.0.9: stdout/stderr threads are now forcefully killed within .1 second if they don't join.\r\n * 1.0.8: Fixes CapturingProcess.kill blocking if the stdout and stderr threads fail to join.\r\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Captures stderr/stdout as a stream to allow easy log monitoring of long running shell processes.\",",
"version": "1.0.13",
"project_urls": {
"Homepage": "https://github.com/zackees/capturing_process"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "b264a41cde56f82360249b3e71950e7a1a96a76b73e96f77c54f4d4fd24f6682",
"md5": "16e9392966601ab6a770c585a73d8fb6",
"sha256": "b2b02c17aad6b304bea31673420db254a9154c9ecf07fcfb1829221cd8fb3eae"
},
"downloads": -1,
"filename": "capturing_process-1.0.13-py2.py3-none-any.whl",
"has_sig": false,
"md5_digest": "16e9392966601ab6a770c585a73d8fb6",
"packagetype": "bdist_wheel",
"python_version": "py2.py3",
"requires_python": ">=3.6.0",
"size": 6609,
"upload_time": "2024-03-25T23:26:49",
"upload_time_iso_8601": "2024-03-25T23:26:49.389916Z",
"url": "https://files.pythonhosted.org/packages/b2/64/a41cde56f82360249b3e71950e7a1a96a76b73e96f77c54f4d4fd24f6682/capturing_process-1.0.13-py2.py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "680cee2ddde6d9e4de7752ce899ee7d9aa836834161086ce47fefb793603a341",
"md5": "2cfd58a15b078b3be4135ae1a783a6b9",
"sha256": "4eb6d2806d68826f62ce509c0253ba46845418c2b75f562e43da37f3d1306a5b"
},
"downloads": -1,
"filename": "capturing_process-1.0.13.tar.gz",
"has_sig": false,
"md5_digest": "2cfd58a15b078b3be4135ae1a783a6b9",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6.0",
"size": 6530,
"upload_time": "2024-03-25T23:26:51",
"upload_time_iso_8601": "2024-03-25T23:26:51.018679Z",
"url": "https://files.pythonhosted.org/packages/68/0c/ee2ddde6d9e4de7752ce899ee7d9aa836834161086ce47fefb793603a341/capturing_process-1.0.13.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-25 23:26:51",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "zackees",
"github_project": "capturing_process",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"tox": true,
"lcname": "capturing-process"
}