ORIONIS API REFERENCE

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

orionis.storage.contracts.IStorageStream

class documentation

class IStorageStream(ABC): (source)

Known subclasses: orionis.storage.AsyncStream

View In Hierarchy

Define the contract for an asynchronous binary stream handle.

Implementations wrap an underlying binary handle (local file, in-memory buffer, remote object) and expose non-blocking read, write, and seek operations. Streams are async context managers:

async with file.open("rb") as stream:
    chunk = await stream.read(65536)
Async Method __aenter__ Open the underlying handle and return the stream.
Async Method __aexit__ Close the stream when leaving the async context.
Async Method close Flush pending data and release the underlying handle.
Async Method read Read up to size bytes from the stream.
Async Method seek Move the stream position to offset.
Async Method write Write data to the stream at the current position.
Class Variable __slots__ Undocumented
@abstractmethod
async def __aenter__(self) -> IStorageStream: (source)
overridden in orionis.storage.AsyncStream

Open the underlying handle and return the stream.

Returns
IStorageStreamThe stream itself, ready for I/O operations.
@abstractmethod
async def __aexit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, traceback: TracebackType | None): (source)
overridden in orionis.storage.AsyncStream

Close the stream when leaving the async context.

Parameters
exc_type:type[BaseException] | NoneException class raised inside the context, if any.
exc:BaseException | NoneException instance raised inside the context, if any.
traceback:TracebackType | NoneTraceback of the raised exception, if any.
Returns
NoneUndocumented
@abstractmethod
async def close(self): (source)
overridden in orionis.storage.AsyncStream

Flush pending data and release the underlying handle.

Returns
NoneUndocumented
@abstractmethod
async def read(self, size: int = -1) -> bytes: (source)
overridden in orionis.storage.AsyncStream

Read up to size bytes from the stream.

Parameters
size:intMaximum number of bytes to read. -1 reads until EOF.
Returns
bytesBytes read from the current position; empty at EOF.
@abstractmethod
async def seek(self, offset: int, whence: int = 0) -> int: (source)
overridden in orionis.storage.AsyncStream

Move the stream position to offset.

Parameters
offset:intTarget offset relative to whence.
whence:intAnchor point: 0 start, 1 current, 2 end.
Returns
intThe new absolute position within the stream.
@abstractmethod
async def write(self, data: bytes) -> int: (source)
overridden in orionis.storage.AsyncStream

Write data to the stream at the current position.

Parameters
data:bytesRaw bytes to write.
Returns
intNumber of bytes written.
__slots__: tuple = (source)
overridden in orionis.storage.AsyncStream

Undocumented