Progress bars¶

ObservabilityLayer emits a TransferEvent per chunk with cumulative transferred bytes. 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:
    payload = b'\0' * (
        64 * 1024 * 1024 * 50
    )  # you produced the source: the total is yours
    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.

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)