Metadata-Version: 2.4
Name: shtoobz
Version: 0.1.0
Summary: Lightweight interchangeable process and TCP byte streams for Windows and Linux.
Author: Jeremy Dunn
Author-email: Jeremy Dunn <jeremy.dunn315@gmail.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Typing :: Typed
Requires-Python: >=3.11
Project-URL: Documentation, https://gitlab.com/stolenfootball-tools/shtoobz/-/blob/main/README.md
Project-URL: Issues, https://gitlab.com/stolenfootball-tools/shtoobz/-/issues
Project-URL: Repository, https://gitlab.com/stolenfootball-tools/shtoobz
Description-Content-Type: text/markdown

# 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

```sh
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

```python
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:

```python
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:

```python
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.

| Method | Behavior |
| --- | --- |
| `recv(n=4096)` | Read up to `n` bytes, waiting only if no data is available. Return `b""` at EOF. |
| `recvn(n)` | Read exactly `n` bytes or raise `ShtoobEOF`/`ShtoobTimeout`. |
| `peek(n=4096)` | Inspect up to `n` bytes without consuming them; wait only if the buffer is empty. |
| `unrecv(data)` | Prepend bytes to the receive buffer. |
| `recvuntil(delimiter)` | Read through and include the delimiter. |
| `recvany(*delimiters)` | Read through the earliest completed delimiter (the shortest matching prefix). |
| `recvline()` | Read through and include the configured newline. |
| `recvall()` | Read until EOF; a short read or idle period is not EOF. |
| `recv_available()` | Collect output until an idle gap or EOF. |
| `iterlines()` | Yield lines and any unterminated final line. |
| `iterchunks(size=4096)` | Yield chunks of at most `size` bytes until EOF. |
| `sendall(data)` | Write all bytes. |
| `sendline(data=b"")` | Write data followed by the configured newline. |
| `sendafter(delimiter, data)` | Receive a prompt, send bytes, and return the prompt. |
| `sendlineafter(delimiter, data=b"")` | Receive a prompt, send a line, and return the prompt. |
| `send_eof()` | Close the sending direction while keeping responses readable. |
| `close()` | Release resources; repeated calls are harmless. |
| `closed` | Whether this shtoob has been closed locally. |
| `write_closed` | Whether the sending direction has been closed locally. |

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.

```python
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:

```python
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:

```python
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:

```python
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.

```python
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.

```python
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.

```python
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:

```python
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:

```python
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](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](https://pypi.org/), using an account with a verified email
   and two-factor authentication enabled.
2. Open [Account settings > Publishing](https://pypi.org/manage/account/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](https://gitlab.com/stolenfootball-tools/shtoobz/-/pipelines).
2. Open [Code > Tags](https://gitlab.com/stolenfootball-tools/shtoobz/-/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](https://pypi.org/project/shtoobz/) 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](https://docs.pypi.org/trusted-publishers/creating-a-project-through-oidc/),
[GitLab tag creation](https://docs.gitlab.com/user/project/repository/tags/#from-the-ui).
