Python wrapper

‹ docs index

processkit-py brings ProcessKit's kernel-backed process-tree containment to Python. It is a thin, typed PyO3 binding over the Rust crate rather than a separate implementation: Job Objects, cgroup v2, POSIX process groups, process I/O, and the teardown state machine remain in Rust.

The Python layer adds the conventions Python applications need:

• matching synchronous and asyncio APIs;
• deterministic with / async with teardown;
• Python exception types with structured fields;
• complete type stubs and a ProcessRunner protocol;
• prebuilt wheels with the optional Rust capabilities already enabled.

Published package: processkit-py on PyPI
Source: ZelAnton/processkit-py
Full guides: processkit-py/docs.

Install

The distribution name is processkit-py; the import name is processkit:

pip install processkit-py

Version 1.0 supports CPython 3.10 and newer. The project publishes abi3 wheels for regular CPython and a version-specific wheel for free-threaded CPython 3.14t. A source distribution is available for platforms without a matching wheel.

The wrapper currently pins Rust processkit 1.2.0 and enables process-control, limits, record, and tracing. Unlike Cargo's optional feature model, every Python wheel exposes one complete API; there are no extras to select at installation time.

Sync and asyncio

Run-to-completion operations have a synchronous name and an a-prefixed asyncio twin. Both return the same result types and use the same Rust runtime semantics:

from processkit import Command

# Non-zero exit is data: inspect code, stdout, stderr, and timed_out.
result = Command("git", ["status", "--short"]).output()

# Require an accepted exit and return trimmed stdout.
version = Command("python", ["--version"]).run()

The same operations compose naturally under asyncio:

import asyncio
from processkit import Command

async def main() -> None:
    result = await Command("git", ["rev-parse", "HEAD"]).aoutput()
    print(result.stdout.strip())

asyncio.run(main())

Cancelling an awaited run drops the underlying Rust future and tears down its owned process tree. Blocking calls release the GIL while Rust waits and check for Python signals, so Ctrl+C can interrupt a synchronous run instead of waiting for the child to finish.

Whole-tree containment

ProcessGroup is a synchronous and asynchronous context manager. Everything started into it shares one containment object and is torn down when the block exits:

from processkit import Command, ProcessGroup

with ProcessGroup() as group:
    group.start(Command("dev-server"))
    group.start(Command("worker"))
    print(group.mechanism)  # job_object | cgroup_v2 | process_group
    print(group.members())

Use the context managers for deterministic cleanup. Windows Job Objects also close when the process loses its last kernel handle. On Unix, cleanup requires Python to execute the teardown path; an uncatchable SIGKILL cannot run it. The POSIX process-group backend also has the same documented setsid escape caveat as the Rust crate. Inspect group.mechanism when the stronger cgroup or Job Object guarantee matters.

Group resource limits, statistics, signals, suspend/resume, kill_all(), and graceful shutdown are exposed directly. Limits require a Windows Job Object or a writable Linux cgroup-v2 root; an unenforceable limit raises ResourceLimit instead of being silently ignored.

Streaming and interactive processes

RunningProcess exposes asyncio-native streaming and stdin. Its result methods consume the handle, matching Rust ownership explicitly: after wait(), finish(), output(), output_bytes(), profile(), or shutdown(), the Python handle is spent.

from processkit import Command

async with await Command("my-build", ["--watch"]).astart() as proc:
    async for line in proc.stdout_lines():
        print(line)

Interactive commands use keep_stdin_open() and take_stdin(). Readiness is provided by the composable Python helpers wait_for, wait_for_line, and wait_for_port.

Pipelines, supervision, and batch runs

The higher-level Rust features have Python-native surfaces:

• Command(...) | Command(...) creates a shell-free Pipeline with pipefail attribution;
• Supervisor keeps a command alive using restart, backoff, and storm-guard policies;
• output_all / aoutput_all run a bounded batch and preserve input order;
• CliClient binds one executable to shared timeout and environment defaults.

Builders are immutable: configuration methods return a new Command, so a base command can be safely reused for several variants.

Errors and results

Like the Rust API, output() treats a non-zero exit and timeout as inspectable result data. Success-checking verbs raise typed exceptions when the outcome is not accepted.

Every package-specific exception derives from ProcessError.

Testing without subprocesses

Application code can depend on the typed ProcessRunner protocol and receive a real Runner in production. Tests can inject the doubles exported from processkit.testing:

• ScriptedRunner returns canned replies and supports scripted streaming;
• RecordingRunner records invocations for assertions;
• RecordReplayRunner records real runs to a cassette and replays them without spawning;
• Reply describes successful, failed, timed-out, signalled, pending, or line-streamed outcomes.

from processkit import Command
from processkit.testing import Reply, ScriptedRunner

runner = ScriptedRunner()
runner.on(["git", "rev-parse"], Reply.ok("abc123\n"))

assert runner.run(Command("git", ["rev-parse", "HEAD"])) == "abc123"

The package ships py.typed and a complete hand-written .pyi surface, checked against the compiled extension with mypy.stubtest.

Where to continue

The Python repository contains the detailed Cookbook, migration guide from subprocess, command reference, streaming guide, and platform matrix.