shtoobz

Lightweight, interchangeable byte streams for Windows and Linux. A shtoob talks to a child process over stdin/stdout or to an application over TCP. The client's communication code stays the same when you switch transports.

Python 3.11+; no third-party runtime dependencies. Managed and built with uv.

Development

uv sync --locked
uv run python -m unittest discover -s tests -v
uv run mypy
uv build

The build produces a wheel and source archive in dist/. To use the checkout from another uv project, run uv add /path/to/shtoobz. No published package is required.

GitLab CI verifies Windows and Linux on Python 3.11 and 3.14. Each job runs strict type checks, builds the distributions, validates their metadata and contents, and runs the integration suite against an isolated wheel installation. The Windows jobs use GitLab's saas-windows-medium-amd64 hosted runner tag; enable hosted runners for the project (or substitute your own Windows runner tag). Branch pipelines save build artifacts. A matching vX.Y.Z tag in the canonical GitLab project publishes to PyPI after all verification jobs pass; see the release steps below.

The source archive includes the tests, lockfile, CI configuration, and MIT license. The installed library has no runtime dependencies; mypy is a development dependency only.

Switch transports

from shtoobz import ProcessShtoob, SocketShtoob

LOCAL = True

shtoob = (
    ProcessShtoob(["./example", "--stdio"], timeout=5)
    if LOCAL
    else SocketShtoob("127.0.0.1", 9000, timeout=5)
)

with shtoob:
    shtoob.recvuntil(b"Name: ")
    shtoob.sendline(b"Alice")
    print(shtoob.recvline())

Use the appropriate executable path on Windows, for example r".\example.exe". Both endpoints must expose the same byte protocol; shtoobz does not translate between a CLI menu and a different network protocol.

A runnable process example on both platforms:

import sys
from shtoobz import ProcessShtoob

with ProcessShtoob(
    [sys.executable, "-u", "-c", "print(input().upper(), flush=True)"],
    timeout=5,
) as shtoob:
    shtoob.sendline(b"hello")
    assert shtoob.recvline().rstrip(b"\r\n") == b"HELLO"

Inject a shtoob into your application instead of constructing a socket inside it:

from shtoobz import Shtoob

class Client:
    def __init__(self, shtoob: Shtoob):
        self.shtoob = shtoob

    def request(self, payload: bytes) -> bytes:
        self.shtoob.sendline(payload)
        return self.shtoob.recvline()

Communication API

All payloads and delimiters are bytes. Encode/decode text explicitly.

Every receive method accepts a keyword timeout. If omitted, the constructor's timeout applies. The default is None (wait indefinitely); an explicit None overrides a finite default. Zero attempts an immediate poll. A finite timeout is a deadline for the read operation, not an idle timeout reset by every chunk. Writes are blocking and do not have a timeout.

Both recv(0) and recvn(0) return b"". Delimiters must be nonempty; negative counts/timeouts and nonfinite timeouts are rejected. Delimiters are included in results, and bytes received after a delimiter are retained for the next read.

On timeout, bytes already received remain buffered. On EOF, an incomplete exact or delimiter read raises ShtoobEOF; its partial attribute contains a snapshot of retained data. Retrieve that data with recv() or recvall(). recvline() therefore raises on an unterminated final line rather than returning it silently.

from shtoobz import ShtoobTimeout

try:
    reply = shtoob.recvuntil(b"> ", timeout=0.5)
except ShtoobTimeout:
    # Continue the same response; previously received bytes have not been lost.
    reply = shtoob.recvuntil(b"> ", timeout=5)

ShtoobTimeout also inherits TimeoutError; ShtoobEOF also inherits EOFError. Operations after local closure raise ShtoobClosed. These all inherit ShtoobError. Other OS failures (connection refused/reset, executable not found, permissions, etc.) retain their standard Python exceptions. A failed write may have sent a prefix; do not retry the entire payload blindly.

Finish input without closing output

Some programs do not produce output until their input stream reaches EOF. send_eof() closes a child's stdin or performs a TCP write shutdown, allowing the same pattern for either transport:

shtoob.sendall(payload)
shtoob.send_eof()
result = shtoob.recvall(timeout=5, max_bytes=1024 * 1024)

Repeated send_eof() calls are harmless while the shtoob remains open. Further writes (including empty writes) raise ShtoobEOF; reads continue normally. After close(), send_eof() raises ShtoobClosed. Half-closing does not terminate a child process. It waits for any ongoing write, so it is not a way to interrupt a blocked writer.

Limit accumulated responses

recvuntil, recvany, recvline, and recvall accept max_bytes=None by default. Pass a nonnegative integer to bound the response, including its delimiter:

from shtoobz import ShtoobLimitExceeded

try:
    line = shtoob.recvline(timeout=5, max_bytes=4096)
except ShtoobLimitExceeded as exc:
    print(f"Response did not fit within {exc.max_bytes} bytes")
    # Received bytes remain available; choose whether to read or close.

A delimiter ending exactly at the limit succeeds. If the limit is reached without a complete delimiter, ShtoobLimitExceeded is raised immediately. A delimiter read with max_bytes=0 therefore always raises. For recvall, exactly the limit is allowed if followed by EOF; one extra byte may be read to distinguish EOF from overflow. That lookahead byte is retained if the operation fails. recvall(max_bytes=0) accepts only an empty stream followed by EOF.

Limits apply to the entire requested response, including data buffered by earlier operations. They never discard previously buffered data. The process reader's separate bounded buffer and OS buffers can still contain additional bytes.

Convenience methods

Prompt helpers return the consumed prompt, including its delimiter:

prompt = shtoob.sendlineafter(b"Name: ", b"Alice", timeout=5, max_bytes=4096)

sendafter() sends the supplied bytes unchanged; sendlineafter() appends the configured newline. Their timeout and byte limit apply to reading the prompt. Writes remain blocking. A failed prompt read sends nothing and preserves partial input. These helpers compose a read and write; they are not atomic transactions against other readers or writers.

peek(n, timeout=...) behaves like recv(n) but leaves the result in the buffer. It can return fewer than n bytes. unrecv(data) prepends bytes, so successive calls put the most recently supplied bytes first. Both work after peer EOF until local closure; unrecv() does not reopen the underlying stream. Neither changes the bytes that have been sent to the application.

prefix = shtoob.peek(4)
header = shtoob.recvn(4)
shtoob.unrecv(header)  # Let another parser consume the complete header.

recv_available(idle_timeout=0.05, timeout=..., max_bytes=None) collects output until no new bytes arrive for idle_timeout seconds or EOF occurs. It returns b"" if that idle period passes without data. Use idle_timeout=0 to drain only bytes already available; a live stream may produce more immediately afterward. The independent timeout is an overall deadline using the constructor's default. It raises ShtoobTimeout and preserves all data if output keeps arriving beyond that deadline. max_bytes behaves like recvall, including its one-byte lookahead. An idle gap is not proof of a complete application message; use a delimiter or known length when the protocol provides one.

output = shtoob.recv_available(idle_timeout=0.1, timeout=2, max_bytes=65536)

iterlines(timeout=..., max_bytes=None) includes each line's newline and yields an unterminated final line at EOF. Its timeout and byte limit apply to each line. iterchunks(size=4096, timeout=...) yields nonempty chunks up to the requested positive size, with a timeout per chunk. A timeout or limit error ends that iterator while leaving incomplete input buffered. Neither iterator holds a read lock between yields, and stopping early leaves the shtoob open and remaining bytes available. Use one logical consumer at a time.

for chunk in shtoob.iterchunks(8192, timeout=5):
    output_file.write(chunk)

Opt-in traffic logging

Pass a standard-library logging.Logger to either constructor. Without one, shtoobz emits no traffic logs, even if the root logger has DEBUG enabled. It never installs handlers or changes logging levels itself:

import logging
from shtoobz import SocketShtoob

logging.basicConfig(level=logging.DEBUG)
with SocketShtoob("localhost", 9000, logger=logging.getLogger("myapp.shtoobz")) as shtoob:
    shtoob.sendline(b"hello")

DEBUG records include the backend class, direction (send or recv), byte count, and the complete payload as a bytes representation. Receive records describe raw chunks entering the shared buffer; peeking, replaying buffered bytes, and unrecv() do not duplicate them. Send records describe successfully completed writes. A failed write may have sent a prefix that is not logged. Exceptions from logging handlers are ignored so diagnostics cannot change successful I/O into an apparent transport failure.

Process options and lifecycle

ProcessShtoob(argv, *, cwd=None, env=None, stderr="inherit", timeout=None, newline=b"\n", buffer_size=65536, logger=None) launches immediately with shell=False. Pass an argument list, not a shell command string. env replaces the child's environment; use {**os.environ, "KEY": "value"} to extend it.

• stderr="inherit" keeps diagnostics out of the protocol stream (default).
• stderr="merge" combines stderr with stdout.
• stderr="discard" ignores stderr.
• newline=b"\r\n" configures both sendline() and recvline() for CRLF.
• pid and returncode expose process status; returncode is None while running.
• wait(timeout=...) waits for normal exit and returns the process exit code.

wait() follows the constructor's timeout by default; wait(None) waits indefinitely and wait(0) polls. Timeout raises ShtoobTimeout without terminating the child. Waiting neither consumes output nor closes the shtoob, and an exited child's status can also be retrieved after close().

The process reader retains its bounded buffer during wait(). A child producing more output than its buffers can hold may block before it exits. Drain output before waiting, or read concurrently; use a finite timeout when output volume is unknown. A typical finite-output flow is:

shtoob.sendall(payload)
shtoob.send_eof()
output = shtoob.recvall(timeout=5, max_bytes=1024 * 1024)
exit_code = shtoob.wait(timeout=5)  # ProcessShtoob only.

A background reader uses a bounded buffer and backpressure for portable timed pipe reads. An individual read still accumulates its response in memory; use max_bytes or read in chunks when handling large or unbounded output. The child must flush output promptly (for Python children, use -u). Bytes are not decoded or newline-translated by shtoobz.

Use a context manager or call close() explicitly. Closing terminates the direct child if still running, waits up to one second, then kills it if necessary and reaps it. It does not terminate descendant processes. Avoid children that leave descendants holding the pipe handles open. A normal EOF does not automatically close the shtoob. Windows console children are launched without a new window.

Socket options

SocketShtoob(host, port, *, timeout=None, connect_timeout=10, newline=b"\n", logger=None) connects immediately to an IPv4 or IPv6 TCP endpoint. connect_timeout is passed to Python's connection helper; DNS resolution is subject to the operating system's resolver behavior. close() shuts down and closes the socket.

Scope

One reader and one writer can operate concurrently. Reads are serialized, as are writes; use one logical consumer so responses are not split between callers. close() wakes blocked readers. No global signal handlers are installed.

This first version supports redirected process pipes and TCP. PTY/ConPTY terminal emulation, interactive terminal forwarding, async APIs, TLS, server listeners, and process-tree management are outside its scope.

When migrating from the original wrappers, change exact recv(n) calls to recvn(n), pass the target to the constructor, and call shtoob.close() instead of accessing an underlying socket.

Licensed under the MIT License. Copyright 2026 Jeremy Dunn.

Publishing to PyPI

Publishing uses GitLab OIDC trusted publishing, following the same approach as ropinator. No PyPI API token is needed. The publish job uses environment pypi and only runs for stable version tags (vX.Y.Z) in stolenfootball-tools/shtoobz. All four Windows/Linux verification jobs must pass first. It uploads the exact wheel and source archive tested by linux: [3.11], without rebuilding them. The release guard rejects a tag/version mismatch, extra or missing files, and distribution metadata that does not match pyproject.toml.

One-time PyPI setup (web UI)

1. Sign in to PyPI, using an account with a verified email and two-factor authentication enabled.

2. Open Account settings > Publishing.

3. Under Add a new pending publisher, select GitLab and enter:

   Field	Value
   PyPI Project Name	shtoobz
   Namespace	stolenfootball-tools
   Project name	shtoobz
   Top-level pipeline file path	.gitlab-ci.yml
   Environment name	pypi

4. Click Add. The publisher stays pending until the first successful upload, which creates the PyPI project. A pending publisher does not reserve the name.

For an existing PyPI project you own, use that project's Manage > Publishing page instead of adding a pending publisher.

Publish version 0.1.0 (web UI)

1. Confirm the latest main pipeline is green under Build > Pipelines.
2. Open Code > Tags and select New tag.
3. Set Tag name to v0.1.0 and Create from to main (or the exact verified release commit). Optionally enter Release 0.1.0 as the message.
4. Select Create tag. This is the action that starts the publishing pipeline.
5. Open the tag's pipeline. After the four verification jobs pass, publish uploads the files automatically. No separate manual job or GitLab release is required. The pypi environment is created by that job automatically.
6. Verify shtoobz on PyPI shows version 0.1.0.

For subsequent releases, update the project version with uv version, commit the updated metadata/lockfile, push, and create the corresponding vX.Y.Z tag once CI passes. Retry a failed publish job only with the same tested artifacts; the upload checks PyPI for already-uploaded identical files. Use a new version for changed release files.

References: PyPI trusted publisher setup, GitLab tag creation.