# running-process
[](../../actions/workflows/lint.yml)
[](../../actions/workflows/push_macos.yml)
[](../../actions/workflows/push_ubuntu.yml)
[](../../actions/workflows/push_win.yml)
A modern subprocess.Popen wrapper with improved process management, real-time output streaming, and enhanced lifecycle control.
## Features
- **Real-time Output Streaming**: Stream process output via queues with customizable formatting
- **Thread-safe Process Management**: Centralized registry for tracking and debugging active processes
- **Enhanced Timeout Handling**: Optional stack trace dumping for debugging hanging processes
- **Process Tree Termination**: Kill entire process trees including child processes (requires psutil)
- **Cross-platform Support**: Works on Windows (MSYS), macOS, and Linux
- **Flexible Output Formatting**: Protocol-based output transformation with built-in formatters
- **Iterator Interface**: Context-managed line-by-line iteration over process output
## Quick Start
### Basic Usage
```python
from running_process import RunningProcess
# Simple command execution with real-time output
process = RunningProcess(["echo", "Hello World"])
for line in process:
print(f"Output: {line}")
# Check exit code
if process.wait() != 0:
print("Command failed!")
```
### Advanced Features
```python
from running_process import RunningProcess
from pathlib import Path
# Advanced configuration
process = RunningProcess(
command=["python", "long_script.py"],
cwd=Path("./scripts"),
timeout=300, # 5 minute timeout
enable_stack_trace=True, # Debug hanging processes
check=True, # Raise exception on non-zero exit
)
# Process output as it arrives
while process.is_running():
try:
line = process.get_next_line(timeout=1.0)
print(f"[{process.elapsed_time:.1f}s] {line}")
except TimeoutError:
print("No output for 1 second...")
continue
# Wait for completion
exit_code = process.wait()
```
### Output Formatting
```python
from running_process import RunningProcess, TimeDeltaFormatter
# Use built-in time delta formatter
formatter = TimeDeltaFormatter()
process = RunningProcess(
["gcc", "-v", "main.c"],
output_formatter=formatter
)
# Implement custom formatter
class TimestampFormatter:
def begin(self): pass
def end(self): pass
def transform(self, line: str) -> str:
from datetime import datetime
timestamp = datetime.now().strftime("%H:%M:%S")
return f"[{timestamp}] {line}"
process = RunningProcess(["make"], output_formatter=TimestampFormatter())
```
### Process Management
```python
from running_process import RunningProcessManager
# Access the global process registry
manager = RunningProcessManager.get_instance()
# List all active processes
for proc_id, process in manager.get_all_processes():
print(f"Process {proc_id}: {process.command_str}")
# Clean up finished processes
manager.cleanup_finished_processes()
```
## Installation
```bash
pip install running_process
```
### Dependencies
This package includes `psutil` as a required dependency for process tree management functionality.
## Architecture
The library follows a layered design with these core components:
- **RunningProcess**: Main class wrapping subprocess.Popen with enhanced features
- **ProcessOutputReader**: Dedicated threaded reader that drains process stdout/stderr
- **RunningProcessManager**: Thread-safe singleton registry for tracking active processes
- **OutputFormatter**: Protocol for transforming process output (with NullOutputFormatter and TimeDeltaFormatter implementations)
- **process_utils**: Utilities for process tree operations
## Development
### Setup
```bash
# Clone the repository
git clone https://github.com/yourusername/running-process.git
cd running-process
# Activate development environment (requires git-bash on Windows)
. ./activate.sh
```
### Testing
```bash
# Run all tests
./test
# Run with coverage
uv run pytest --cov=running_process tests/
```
### Linting
```bash
# Run complete linting suite
./lint
# Individual tools
uv run ruff check --fix src tests
uv run black src tests
uv run pyright src tests
```
## API Reference
### RunningProcess
The main class for managing subprocess execution:
```python
class RunningProcess:
def __init__(
self,
command: str | list[str],
cwd: Path | None = None,
check: bool = False,
auto_run: bool = True,
shell: bool | None = None,
timeout: int | None = None,
enable_stack_trace: bool = False,
on_complete: Callable[[], None] | None = None,
output_formatter: OutputFormatter | None = None,
) -> None: ...
def get_next_line(self, timeout: float | None = None) -> str | EndOfStream: ...
def wait(self, timeout: float | None = None) -> int: ...
def kill(self) -> None: ...
def is_running(self) -> bool: ...
def drain_stdout(self) -> list[str]: ...
```
### Key Methods
- `get_next_line(timeout)`: Get the next line of output with optional timeout
- `wait(timeout)`: Wait for process completion, returns exit code
- `kill()`: Terminate the process (and process tree if psutil available)
- `is_running()`: Check if process is still executing
- `drain_stdout()`: Get all currently available output lines
### ProcessOutputReader
Internal threaded reader that drains process stdout/stderr:
```python
class ProcessOutputReader:
def __init__(
self,
proc: subprocess.Popen[Any],
shutdown: threading.Event,
output_formatter: OutputFormatter | None,
on_output: Callable[[str | EndOfStream], None],
on_end: Callable[[], None],
) -> None: ...
def run(self) -> None: ... # Thread entry point
```
### OutputFormatter Protocol
```python
class OutputFormatter(Protocol):
def begin(self) -> None: ...
def transform(self, line: str) -> str: ...
def end(self) -> None: ...
```
Built-in implementations:
- `NullOutputFormatter`: No-op formatter (default)
- `TimeDeltaFormatter`: Adds elapsed time prefix to each line
## License
BSD 3-Clause License
## Contributing
1. Fork the repository
2. Create a feature branch
3. Make your changes following the existing code style
4. Run tests and linting: `./test && ./lint`
5. Submit a pull request
For bug reports and feature requests, please use the [GitHub Issues](https://github.com/yourusername/running-process/issues) page.
Raw data
{
"_id": null,
"home_page": null,
"name": "running-process",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "cli, management, process, subprocess",
"author": null,
"author_email": "Your Name <your.email@example.com>",
"download_url": null,
"platform": null,
"description": "# running-process\n\n[](../../actions/workflows/lint.yml)\n[](../../actions/workflows/push_macos.yml)\n[](../../actions/workflows/push_ubuntu.yml)\n[](../../actions/workflows/push_win.yml)\n\nA modern subprocess.Popen wrapper with improved process management, real-time output streaming, and enhanced lifecycle control.\n\n## Features\n\n- **Real-time Output Streaming**: Stream process output via queues with customizable formatting\n- **Thread-safe Process Management**: Centralized registry for tracking and debugging active processes\n- **Enhanced Timeout Handling**: Optional stack trace dumping for debugging hanging processes\n- **Process Tree Termination**: Kill entire process trees including child processes (requires psutil)\n- **Cross-platform Support**: Works on Windows (MSYS), macOS, and Linux\n- **Flexible Output Formatting**: Protocol-based output transformation with built-in formatters\n- **Iterator Interface**: Context-managed line-by-line iteration over process output\n\n## Quick Start\n\n### Basic Usage\n\n```python\nfrom running_process import RunningProcess\n\n# Simple command execution with real-time output\nprocess = RunningProcess([\"echo\", \"Hello World\"])\nfor line in process:\n print(f\"Output: {line}\")\n\n# Check exit code\nif process.wait() != 0:\n print(\"Command failed!\")\n```\n\n### Advanced Features\n\n```python\nfrom running_process import RunningProcess\nfrom pathlib import Path\n\n# Advanced configuration\nprocess = RunningProcess(\n command=[\"python\", \"long_script.py\"],\n cwd=Path(\"./scripts\"),\n timeout=300, # 5 minute timeout\n enable_stack_trace=True, # Debug hanging processes\n check=True, # Raise exception on non-zero exit\n)\n\n# Process output as it arrives\nwhile process.is_running():\n try:\n line = process.get_next_line(timeout=1.0)\n print(f\"[{process.elapsed_time:.1f}s] {line}\")\n except TimeoutError:\n print(\"No output for 1 second...\")\n continue\n\n# Wait for completion\nexit_code = process.wait()\n```\n\n### Output Formatting\n\n```python\nfrom running_process import RunningProcess, TimeDeltaFormatter\n\n# Use built-in time delta formatter\nformatter = TimeDeltaFormatter()\nprocess = RunningProcess(\n [\"gcc\", \"-v\", \"main.c\"],\n output_formatter=formatter\n)\n\n# Implement custom formatter\nclass TimestampFormatter:\n def begin(self): pass\n def end(self): pass\n\n def transform(self, line: str) -> str:\n from datetime import datetime\n timestamp = datetime.now().strftime(\"%H:%M:%S\")\n return f\"[{timestamp}] {line}\"\n\nprocess = RunningProcess([\"make\"], output_formatter=TimestampFormatter())\n```\n\n### Process Management\n\n```python\nfrom running_process import RunningProcessManager\n\n# Access the global process registry\nmanager = RunningProcessManager.get_instance()\n\n# List all active processes\nfor proc_id, process in manager.get_all_processes():\n print(f\"Process {proc_id}: {process.command_str}\")\n\n# Clean up finished processes\nmanager.cleanup_finished_processes()\n```\n\n## Installation\n\n```bash\npip install running_process\n```\n\n### Dependencies\n\nThis package includes `psutil` as a required dependency for process tree management functionality.\n\n## Architecture\n\nThe library follows a layered design with these core components:\n\n- **RunningProcess**: Main class wrapping subprocess.Popen with enhanced features\n- **ProcessOutputReader**: Dedicated threaded reader that drains process stdout/stderr\n- **RunningProcessManager**: Thread-safe singleton registry for tracking active processes\n- **OutputFormatter**: Protocol for transforming process output (with NullOutputFormatter and TimeDeltaFormatter implementations)\n- **process_utils**: Utilities for process tree operations\n\n## Development\n\n### Setup\n\n```bash\n# Clone the repository\ngit clone https://github.com/yourusername/running-process.git\ncd running-process\n\n# Activate development environment (requires git-bash on Windows)\n. ./activate.sh\n```\n\n### Testing\n\n```bash\n# Run all tests\n./test\n\n# Run with coverage\nuv run pytest --cov=running_process tests/\n```\n\n### Linting\n\n```bash\n# Run complete linting suite\n./lint\n\n# Individual tools\nuv run ruff check --fix src tests\nuv run black src tests\nuv run pyright src tests\n```\n\n## API Reference\n\n### RunningProcess\n\nThe main class for managing subprocess execution:\n\n```python\nclass RunningProcess:\n def __init__(\n self,\n command: str | list[str],\n cwd: Path | None = None,\n check: bool = False,\n auto_run: bool = True,\n shell: bool | None = None,\n timeout: int | None = None,\n enable_stack_trace: bool = False,\n on_complete: Callable[[], None] | None = None,\n output_formatter: OutputFormatter | None = None,\n ) -> None: ...\n\n def get_next_line(self, timeout: float | None = None) -> str | EndOfStream: ...\n def wait(self, timeout: float | None = None) -> int: ...\n def kill(self) -> None: ...\n def is_running(self) -> bool: ...\n def drain_stdout(self) -> list[str]: ...\n```\n\n### Key Methods\n\n- `get_next_line(timeout)`: Get the next line of output with optional timeout\n- `wait(timeout)`: Wait for process completion, returns exit code\n- `kill()`: Terminate the process (and process tree if psutil available)\n- `is_running()`: Check if process is still executing\n- `drain_stdout()`: Get all currently available output lines\n\n### ProcessOutputReader\n\nInternal threaded reader that drains process stdout/stderr:\n\n```python\nclass ProcessOutputReader:\n def __init__(\n self,\n proc: subprocess.Popen[Any],\n shutdown: threading.Event,\n output_formatter: OutputFormatter | None,\n on_output: Callable[[str | EndOfStream], None],\n on_end: Callable[[], None],\n ) -> None: ...\n\n def run(self) -> None: ... # Thread entry point\n```\n\n### OutputFormatter Protocol\n\n```python\nclass OutputFormatter(Protocol):\n def begin(self) -> None: ...\n def transform(self, line: str) -> str: ...\n def end(self) -> None: ...\n```\n\nBuilt-in implementations:\n- `NullOutputFormatter`: No-op formatter (default)\n- `TimeDeltaFormatter`: Adds elapsed time prefix to each line\n\n## License\n\nBSD 3-Clause License\n\n## Contributing\n\n1. Fork the repository\n2. Create a feature branch\n3. Make your changes following the existing code style\n4. Run tests and linting: `./test && ./lint`\n5. Submit a pull request\n\nFor bug reports and feature requests, please use the [GitHub Issues](https://github.com/yourusername/running-process/issues) page.",
"bugtrack_url": null,
"license": "BSD 3-Clause License",
"summary": "A modern subprocess.Popen wrapper with improved process management",
"version": "1.0.6",
"project_urls": {
"Homepage": "https://github.com/yourusername/running-process",
"Issues": "https://github.com/yourusername/running-process/issues",
"Repository": "https://github.com/yourusername/running-process"
},
"split_keywords": [
"cli",
" management",
" process",
" subprocess"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "dc59eb602f06cd6ba119f26ff3007afbff4e727dd9c6742b0cf7962484886d2e",
"md5": "7a252f899a476f6814f629afb88924f4",
"sha256": "fb77250ab05ebbd5e727da4ea5d0555b0109ffec573d31e1f952612b346253cd"
},
"downloads": -1,
"filename": "running_process-1.0.6-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7a252f899a476f6814f629afb88924f4",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 27275,
"upload_time": "2025-10-08T11:44:39",
"upload_time_iso_8601": "2025-10-08T11:44:39.666332Z",
"url": "https://files.pythonhosted.org/packages/dc/59/eb602f06cd6ba119f26ff3007afbff4e727dd9c6742b0cf7962484886d2e/running_process-1.0.6-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-08 11:44:39",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "yourusername",
"github_project": "running-process",
"github_not_found": true,
"lcname": "running-process"
}
JSON
