Python streams and mappings

Use byte streams for generated data or incremental reads, and mappings to choose each entry’s source and destination. See the Python guide for examples and the operations reference for client configuration and shared arguments. Results and exceptions apply across these interfaces.

Byte streams

client.open_writer(*, as_=None, as_new=None, as_existing=None, to=None, follow_dst=False, ...) returns a StreamWriter. Choose exactly one of as_, as_new, or as_existing; the latter two require the destination to be absent or present, following cp placement semantics. Writers have no source basename, so they require an exact destination path. client.open_reader(src, *, from_=None, cwd=None, root=None, follow_src=False, ...) returns a StreamReader. cwd resolves relative sources; root also confines them. Choose at most one, as with cp. These bases belong to the source endpoint, independently of the client’s local process_cwd. Both accept rsh, syq_path, pscope, no_bootstrap, no_compress, no_tcp, tcp_plain, tcp_ports, tcp_congestion, auth_from (S3), s3_endpoint, s3_region, s3_profile, s3_header, performance_tuning, resource_limits, integrity_checking, only_new, dry_run, stats, verbose, quiet, progress, no_progress, and timeout with the same meanings as cp. See the CLI stream reference for the applicable tuning and integrity controls. The client supplies the executable, process working directory, environment, and default timeout. Stream calls always check transfer failures.

These methods transfer raw bytes, using the CLI’s stream semantics for destination permissions, metadata, and endpoint restrictions. They are sequential, non-seekable interfaces; use one operation at a time on each stream. No whole-object retry or restart recovery is attempted. S3 can retry buffered multipart parts.

Streams use bounded transport buffers. read() without a size collects all remaining bytes in Python memory and checks completion before returning.

Writer completion

Use a with block to commit on successful exit or abort on an exception. An abort before commit does not replace the destination. close() only ends payload input; it does not publish the file. Successful context exit commits even if you already called close(), unless you also called abort().

Writers support io.BufferedWriter and io.TextIOWrapper. Close the wrapper before committing the underlying writer so its buffered data is flushed.

Outside a context, finish with commit() or abort(); close() alone leaves the transfer pending. commit() can follow payload closure, is repeatable after success, and fails after an abort. Explicit commit publishes immediately, so later application errors cannot undo it. A timeout or connection loss during commit can leave the outcome uncertain; the method reports failure rather than claiming rollback. Whole datasets need their own final publication step after all object transfers succeed.

Skips and previews

Writers normally return while destination setup continues, so opening several writers lets their connections start concurrently. Setup errors can surface at write() or commit(). With only_new=True or dry_run=True, opening waits for the destination decision (also for options inherited from the environment). Check output.skipped before producing data:

with client.open_writer(to="server", as_="archive.tar", only_new=True) as output:
    if not output.skipped:
        produce_archive(output)

A skipped writer rejects write() and exits its context successfully without committing anything. Skipped readers return no payload; their skipped property is settled at completion. Both count the skipped object in files_excluded.

With dry_run=True, a context checks placement without transferring bytes. Enter and exit a writer context without calling write(); preview writers reject payload writes. Preview readers return no payload. The result contains planned totals; bytes_total_known=False distinguishes an unknown pipe length from an empty source. A dry run does not check an expected payload hash.

Results and failures

After completion, stream.result holds a CpResult, including byte counts and elapsed time. It remains None if no terminal result arrived, such as after forced termination. Missing or invalid completion records raise SyqProtocolError even if the process exits successfully.

A bounded reader call can yield partial data before a later transfer error. An unbounded read() checks transfer completion before returning its bytes. An exception inside either context cancels the transfer and preserves the original exception. Writers that have not committed are aborted during garbage collection, including after payload closure; cleanup can block, so use explicit contexts for timely cleanup.

timeout covers the stream’s lifetime, including blocked reads/writes. Timeout raises subprocess.TimeoutExpired unless the transfer has completed successfully; transfer failures raise SyqProcessError, whose result contains the exit status and the last 8 KiB of diagnostics, without capturing payload bytes. The stream’s stderr property exposes those same last 8 KiB as bytes, including on success. For example, request stats=True and read output.stderr.decode() after the writer context exits.

Async streams

Async streams use async with directly and await I/O and explicit closure:

client = syq.AsyncClient()
async with client.open_writer(to="server", as_="generated.bin") as output:
    await output.write(b"generated bytes")
async with client.open_reader("generated.bin", from_="server") as source:
    while chunk := await source.read(65536):
        await consume(chunk)

Async streams start on context entry. Cancellation terminates the owned transfer process and releases blocked I/O. AsyncStreamReader and AsyncStreamWriter expose async read/write, close, and abort methods. AsyncStreamWriter.commit() explicitly publishes, with the same semantics as its synchronous counterpart; successful async context exit commits automatically.

map

map(*sources, **options) → MapStream lists local mapping entries without copying. Besides the shared arguments, it accepts as_ to rename a selected object. srcs_in must be the sole selector when used.

MapStream is a Mapping and an iterable context manager; use with. AsyncMapStream is an AsyncMapping and an async context manager; use async with. Streams must be consumed inside their context and cannot be reused after completion or closure. Async streams capture the producer directory, environment, and timeout when created, even though execution starts later.

Mapping and AsyncMapping

A mapping combines entries with their local source context. Pass it directly to cp(mapping=...), even on a client with a different process_cwd.

Mapping(entries, *, cwd=None, root=None, follow_src=False) accepts an iterable of MappingEntry. AsyncMapping(...) accepts an async iterable. If both cwd and root are omitted, the source base is the current directory at construction. Otherwise exactly one may be supplied. Relative cwd and root paths resolve against the Python process directory at construction, not a client’s process_cwd. root confines source resolution.

The context properties are read-only. A transform receives each MappingEntry and returns a replacement entry, or None to omit it. Transformations can be chained. Async transforms also accept awaitable callbacks and await them in entry order. Neither form caches entries; reuse depends on the supplied iterable.

A context-carrying mapping rejects from_, cwd, and root overrides on cp. It automatically enables the source-following policy used by map; enabling follow or follow_src on the copy is also permitted. The destination options remain independent. map(root=..., srcs_in=...) carries the selected directory as the consuming copy’s root. The copy resolves that root again; it does not inherit an open directory handle or a snapshot of the source tree.

Normal end of stream iteration checks the mapping process status. Leaving its context early stops the process. An exhausted or closed stream cannot be copied as an empty mapping.

For cp(mapping=iterable), the entire iterable is saved to a temporary manifest before copying starts. An iteration, transformation, or serialization failure starts no copy. Plain iterables and manifest paths have no source context; pass cwd, root, or from_ explicitly as needed. See mapping rules.

Callback mappings

StreamSource(produce, *, size=None) supplies bytes to one mapping entry. StreamDestination(consume) receives them. Pass either or both as the src and dst of a MappingEntry; callback entries require kind=None or "file". These objects work with live cp(mapping=...) calls, not saved pathname manifests.

The complete iterable is validated before any callback or destination change. Callbacks start lazily, with at most stream_concurrency entries active; one entry can have both a producer and consumer. Pathname entries run first with their usual comparison and recovery behavior, then callback entries share transfer sessions. A mixed mapping therefore has one endpoint setup per phase. A failed entry does not undo entries that already succeeded.

Producers receive a writable binary file object. Returning normally authorizes publication; closing the object only ends the bytes. An exception, including one raised after a wrapper closes the writer, cancels unpublished work. Consumers receive a readable binary object. Reading through EOF checks transfer success; a normal early return drains and checks the rest. Consumer exceptions cancel remaining work. Write extracted files into a staging directory and publish it after cp succeeds. syq cannot undo a consumer’s own side effects.

Client runs callbacks on worker threads. AsyncClient accepts the same synchronous callbacks, or async def callbacks that await read, write, and close; async callbacks run on the caller’s event loop. Cancellation releases blocked payload I/O. Application code that keeps computing without I/O must cooperate with cancellation. Do not retain the supplied file object after return.

StreamSource.size promises the exact byte count; a mismatch fails before publication. MappingEntry.size remains informational. expected_hash checks the bytes during transfer on every backend; it adds no second download. Without an expectation or requested integrity check, no extra whole-stream hash is made. metadata sets attributes on named destinations without preserve. Source callbacks have no file metadata to preserve. S3 stores explicit attributes in syq’s existing object metadata format; omitted attributes use new-file defaults.

Skip policies and dry runs do not invoke excluded callbacks. Callbacks always transfer when selected; they have no saved content identity for comparison or restart recovery. They cannot use hash, content comparison, inplace, pruning, or source-tree ignore rules. Filter entries in your script. Named receiver grants authorize paths and do not accept callbacks. Callbacks run locally; mixed pathname transfers between SSH hosts require coordinate_at="local".

Callback runs emit MappingStreamResult events with the original zero-based entry index, source and destination (PathValue, or None for a callback), disposition, dry_run, optional bytes, and optional error message. After a completed run, use failed entry indices to construct an application retry. Callbacks are never replayed automatically. Their automation records use schema version 3; ordinary pathname and descriptor calls retain version 2.

Hash and HashAlgorithm

Hash(algorithm, value) describes the expected hash of all bytes in one regular file. algorithm accepts a HashAlgorithm value or its string: "blake3", "sha256", "md5", or "xxh3-128". value is hexadecimal: 64 digits for BLAKE3 and SHA-256, 32 for MD5 and XXH3-128. The immutable object validates the length and characters and stores lowercase hex.

The expectation covers the complete resulting file, including reused bytes. A mismatch fails the file rather than reporting a successful copy. Files excluded by selection rules are not hash-verified. hash=True still controls whether existing contents are compared instead of trusting size and modification time. Dry runs preview changes without validating the expectation. An expected whole-file hash is independent of the algorithm used for block comparison or transport checks. MD5 and XXH3-128 are useful for compatibility and accidental-error detection, but do not provide cryptographic collision resistance.

MappingEntry

Frozen dataclass describing one source-to-destination mapping. Pass an iterable of these to cp(mapping=...); use dataclasses.replace to change an entry.

MappingEntry(src, dst, kind=None, size=None, mtime=None, expected_hash=None, metadata=None) also accepts text or byte paths for src and dst and converts them to RelativePath. size and mtime do not impose preconditions on the copy. expected_hash does: a file cannot succeed unless its contents match. For example, an adapter can supply an MD5 from a DVC manifest without changing syq’s ordinary comparison algorithm:

entry = syq.MappingEntry(
    "cache/object", "data.bin", kind="file",
    expected_hash=syq.Hash("md5", "900150983cd24fb0d6963f7d28e17f72"),
)
client.cp(mapping=[entry], cwd="source", into="download")

Mapping files encode it as "expected_hash": {"algorithm": "md5", "value": "..."}. Older syq versions that do not support this field reject the mapping.

DestinationMetadata

Set destination attributes without modifying the source:

entry = syq.MappingEntry(
    "source.bin", "payload.bin",
    metadata=syq.DestinationMetadata(mode=0o640, mtime=1700000000),
)
client.cp(mapping=[entry], cwd="source", into="output")

The keyword-only fields are mode, uid, gid, mtime, and mtime_nsec, all optional integers. mode contains permission bits only; uid and gid are numeric IDs. Times use Unix seconds plus optional nanoseconds. Supplying mtime without mtime_nsec uses zero nanoseconds. Omitted attributes follow normal copy behavior. Explicit attributes do not require preserve, except where a restricted receiver’s signed grant needs the corresponding permission. The mapping reference describes backend behavior.

RelativePath and PathValue

RelativePath(value) accepts text, bytes, or os.PathLike. It rejects empty paths, absolute paths, NUL bytes, and empty, . or .. components. Join paths with /, for example syq.RelativePath("archive") / entry.dst.

PathValue(raw: bytes) holds a path received in an event, which may be absolute. Both types are immutable and provide:

RelativePath also implements os.PathLike, returning bytes.