Install & Compatibility
Where this runs
tested against v1.0.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.920 runs
installs and imports cleanly · install 0.0s · import 0.217s · 17.8MB
glibcpy 3.10–3.920 runs
installs and imports cleanly · install 1.5s · import 0.186s · 18MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
SupervisedProcess
✓ from simpervisor import SupervisedProcess
This example demonstrates how to use `SupervisedProcess` to start an asynchronous Python script as a child process, monitor its status with `ready()`, and gracefully terminate it using `terminate()`. The child process simulates some work before exiting or being cancelled. The `sys.executable` is used to ensure the same Python interpreter runs the child script.
import asyncio
import sys
import os
from simpervisor import SupervisedProcess
# Create a dummy script to supervise
dummy_script_content = """
import asyncio
import sys
import time
async def child_process_task():
print(f"Child process {os.getpid()} started.")
try:
for i in range(5):
print(f"Child process {os.getpid()}: Working... {i+1}/5")
await asyncio.sleep(1)
print(f"Child process {os.getpid()}: Done working, exiting.")
except asyncio.CancelledError:
print(f"Child process {os.getpid()}: Cancelled, cleaning up.")
except Exception as e:
print(f"Child process {os.getpid()}: Error - {e}")
finally:
print(f"Child process {os.getpid()} exiting gracefully.")
if __name__ == '__main__':
asyncio.run(child_process_task())
"""
async def main():
script_path = "./dummy_child_script.py"
with open(script_path, "w") as f:
f.write(dummy_script_content)
print("Starting supervisor example...")
# Supervise the dummy script using python as the executable
# This assumes 'python' is in your PATH
process = SupervisedProcess(
[sys.executable, script_path], # Command to run the child process
always_restart=False # For this example, don't restart automatically
)
await process.start()
print(f"Supervisor: Process started with PID {process.pid}")
await asyncio.sleep(2) # Give child some time to work
if await process.ready():
print("Supervisor: Child process reports ready (or has started).")
else:
print("Supervisor: Child process not yet ready.")
await asyncio.sleep(3)
print("Supervisor: Attempting to terminate child process...")
await process.terminate()
# await process.wait() # Can wait for termination if needed
if not process.running:
print("Supervisor: Child process terminated.")
else:
print("Supervisor: Child process is still running after terminate attempt.")
os.remove(script_path)
print("Supervisor: Example finished.")
if __name__ == '__main__':
asyncio.run(main())
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'simpervisor'
The `simpervisor` package has not been installed in the current Python environment.
fixpip install simpervisor
RuntimeWarning: coroutine 'SupervisedProcess.start' was never awaited
An asynchronous method of `SupervisedProcess` was called without the `await` keyword, meaning the coroutine was created but not executed.
fixPrefix the method call with `await`, for example: `await process.start()`.
ImportError: cannot import name 'start' from 'simpervisor'
`start` is an instance method of the `SupervisedProcess` class, not a top-level function importable directly from the `simpervisor` module.
fixImport `SupervisedProcess` and call `start` on an instance: `from simpervisor import SupervisedProcess; p = SupervisedProcess(...); await p.start()`.
FileNotFoundError: [Errno 2] No such file or directory: 'your_command_here'
The executable specified in the `command` list for `SupervisedProcess` could not be found by the operating system.
fixVerify the command's spelling and ensure it is in your system's PATH or provide its full path.
AttributeError: 'SupervisedProcess' object has no attribute 'stdout'
`SupervisedProcess` does not directly expose `stdout`, `stdin`, or `stderr` attributes; it uses `stdout_logs` and `stderr_logs` queues for output management.
fixAccess output using `await process.stdout_logs.get()` or `await process.stderr_logs.get()`, ensuring `capture_output=True` in the constructor.
Upgrade
Version history
1.0.0latest on PyPI · released May 18, 2023
Audit
Dependencies
PythonrequiredRequired runtime environment