procwire

procwire

Process lifecycle orchestration and task piping for Python services. Manage spawned processes with tracking, timeout control, and event-driven lifecycle hooks.

Installation

pip install procwire

Usage

import procwire

# Spawn a tracked background process
worker = procwire.spawn('my-worker', 'python', ['-m', 'myapp.worker'], timeout=30000)
print(f"Worker PID: {worker['pid']}")

# Check process status
info = procwire.status('my-worker')
print(f"Running: {info['running']}, Uptime: {info['uptime']:.1f}s")

# Run synchronous command with captured output
result = procwire.run_sync('git', ['status', '--short'], timeout=10)
print(result['stdout'])

# Lifecycle events
procwire.on('exit', lambda name, code: print(f'{name} exited: {code}'))
procwire.on('timeout', lambda name, ms: print(f'{name} timed out after {ms}ms'))

# Restart crashed workers
procwire.restart('my-worker', 'python', ['-m', 'myapp.worker'])

# Graceful shutdown
procwire.cleanup()

API

Process Management

• spawn(name, cmd, args=None, timeout=None, detached=True, hide=True) — spawn tracked process
• spawn_shell(name, command, timeout=None) — spawn via shell
• run_sync(cmd, args=None, timeout=30, capture=True) — run and wait
• kill(name, sig=None) — terminate tracked process
• restart(name, cmd=None, args=None, **kwargs) — kill and respawn

Status

• list_processes() — list tracked names
• has(name) — check if tracked
• status(name) — get status dict
• status_all() — all statuses
• is_running(name) / pid_of(name) / uptime(name)

Lifecycle

• cleanup() — kill all tracked processes
• wait_all(timeout=None) — wait for all to exit
• on(event, fn) / once(event, fn) / off(event, fn)

Events

• spawn — process started
• exit — process exited
• timeout — process timed out
• kill — process killed

License

MIT