Progress bars¶

ObservabilityLayer emits a TransferEvent per chunk with cumulative transferred bytes for that stream. storix never guesses a total: you produced the source, so the total is yours (a local file's stat().st_size, an HTTP upload's Content-Length), and a percentage is your division to make. That one contract drives any UI: a bar when you have a total, a counter when you do not.

Drive a rich bar¶

You own the payload, so you own the total; the sink only moves the bar:

uv add rich

"""Upload and download progress bars from ObservabilityLayer transfer events."""

from typing import Final

from rich.progress import Progress

from storix import ObservabilityLayer, TransferEvent, get_storage


CHUNK_SIZE: Final[int] = 64 * 1024
"""Transfer chunk size; one TransferEvent fires per chunk."""


def main() -> None:
    # 64 MiB: 1024 chunk events per bar, gentle on RAM. Multiply by 50
    # for a slow-motion bar - but note the demo materializes its whole
    # source in memory first (~3 GiB then), purely to keep the example
    # short; storix itself streams chunk by chunk either way.
    payload = b'\0' * (64 * 1024 * 1024)
    chunks = (payload[i : i + CHUNK_SIZE] for i in range(0, len(payload), CHUNK_SIZE))

    with Progress() as progress:
        tasks = {
            'write': progress.add_task('uploading', total=len(payload)),
            'read': progress.add_task('downloading', total=len(payload)),
        }

        def on_event(event: TransferEvent) -> None:
            progress.update(tasks[event.op], completed=event.transferred)

        fs = get_storage('memory').with_layer(ObservabilityLayer, sink=on_event)
        fs.echo(chunks, '/upload.bin')
        for _chunk in fs.stream('/upload.bin', chunk_size=CHUNK_SIZE):
            pass  # hand each chunk to its real consumer here


if __name__ == '__main__':
    main()

Attach the layer outermost (the last with_layer) so it counts the full logical transfer, whatever other layers sit below it.

One file, several streams¶

transferred is cumulative per stream, and a parallel download() reads one file through several ranges at once, each counting from zero. The event carries offset - where its stream starts in the file - so a sink that accumulates a whole transfer keys on the pair:

seen: dict[tuple[PurePosixPath, int], int] = {}
completed = 0


def on_event(event: TransferEvent) -> None:
    global completed
    stream = (event.path, event.offset)
    completed += event.transferred - seen.get(stream, 0)
    seen[stream] = event.transferred

Keying on path alone silently undercounts: with eight ranges in flight the bar reports one range's bytes rather than the file's. offset is 0 for a whole-file stream() or echo(), so a sink written this way is correct for every transfer, sequential or parallel.

Stop a transfer from the sink¶

The sink runs once per chunk, in whichever thread produced it, which makes it the one place a transfer can be stopped from. Raise, and the exception unwinds that stream; with a bulk transfer running several files at once, the others stop at their own next chunk:

import threading

from storix.errors import TransferStoppedError

stop = threading.Event()


def on_event(event: TransferEvent) -> None:
    if stop.is_set():
        raise TransferStoppedError
    bar.update(event.transferred)

TransferStoppedError is deliberately not a StorageError - nothing failed - and it derives from BaseException, the same choice the standard library makes for asyncio.CancelledError. A stop request must not be swallowed by an except Exception between your callback and your code: in a custom layer, a third-party backend, or a provider SDK. So catch it by name:

try:
    fs.download("/media/movie.mkv", sink)
except TransferStoppedError:
    partial.unlink(missing_ok=True)

Any exception you raise from a sink will unwind the stream - that is just how generators work - but only this one is guaranteed to reach you intact.

This is exactly how sx implements Ctrl+C: the signal handler sets an event instead of raising, and the sink turns it into TransferStoppedError inside every worker. A partially written destination is the caller's to clean up - storix does not guess whether a half-file is worth keeping.

No total? Count bytes¶

A truly unbounded source (a generator, a pipe) has no end to know. Render a counter or a spinner instead of a bar; transferred is still exact:

from storix import ObservabilityLayer, TransferEvent, get_storage


def on_event(event: TransferEvent) -> None:
    print(f"\r{event.op} {event.path}: {event.transferred:,} bytes", end="")


fs = get_storage("local").with_layer(ObservabilityLayer, sink=on_event)

Async sinks¶

With storix.aio the sink may be a coroutine function (push SSE frames, websocket messages, ...); it is awaited once per chunk:

async def on_event(event: TransferEvent) -> None:
    await queue.put(event)