bytes_buffer.py

BytesBuffer dataclass

Bases: IO[bytes]

IO that uses an in-memory bytes buffer to load and save data. Useful for buffering data across nodes without writing to disk.

Example:

>>> from ordeq_common import BytesBuffer
>>> buffer = BytesBuffer()
>>> buffer.load()
b''

The buffer is initially empty, unless provided with initial data:

>>> buffer = BytesBuffer(b"Initial data")
>>> buffer.load()
b'Initial data'

Saving to the buffer appends data to the existing content:

>>> buffer.save(b"New data")
>>> buffer.load()
b'Initial dataNew data'

Example in a node:

>>> from ordeq_args import CommandLineArg
>>> from ordeq_common import BytesBuffer, Literal
>>> from ordeq import node, run
>>> result = BytesBuffer(b"Greeting")
>>> @node(
...     inputs=[BytesBuffer(b"Hello"), Literal(b"you")], outputs=result
... )
... def greet(greeting: bytes, name: bytes) -> bytes:
...     return greeting + b", " + name + b"!"
>>> run(greet).get(result)
b'Hello, you!'