io

BufferedBinaryIterator

Iterator for sending bytes as chunks.

Its object can be used to generate iterator from a bytes object, which yields specified size of bytes objects.

Note

Even if you generate its object from a bytes object sending to a client and specify it as response body, there may be no effect on performance of your web application. Its object should be used only for unifying interface as a subclass of ResponseBodyIteratorBase.

__init__(self, data, bufsize=8192) special

Parameters:

Name	Type	Description	Default
data	bytes	A bytes object to be iterated.	required
bufsize	int	Chunk size of bytes objects yielded from the iterator.	8192

Source code in bamboo/io.py

def __init__(self, data: bytes, bufsize: int = 8192) -> None:
    """
    Args:
        data: A bytes object to be iterated.
        bufsize: Chunk size of bytes objects yielded from the iterator.
    """
    super().__init__()

    self._data = io.BytesIO(data)
    self._bufsize = bufsize

BufferedConcatIterator

Iterator conbining serveral iterator yeilding bytes objects.

Its object can concatenate several iterator or bytes objects and behaves as a single iterator yieding bytes objects with specified chunk size.

__init__(self, *items, *, bufsize=8192) special

Parameters:

Name	Type	Description	Default
*items	Union[bytes, Iterator[bytes]]	bytes objects or iterators yielding bytes.	()
bufsize	int	Chunk size of bytes objects yielded from the iterator.	8192

Source code in bamboo/io.py

def __init__(
    self,
    *items: t.Union[bytes, t.Iterator[bytes]],
    bufsize: int = 8192
) -> None:
    """
    Args:
        *items: bytes objects or iterators yielding bytes.
        bufsize: Chunk size of bytes objects yielded from the iterator.
    """
    super().__init__()

    self._iters: t.Deque[t.Iterator[bytes]] = deque()
    self._current: t.Iterator[bytes] = None
    self._bufsize = bufsize
    self._buffer = io.BytesIO()

    for item in items:
        self.append(item)

append(self, item)

Add bytes or iterator yielding bytes inside.

Parameters:

Name	Type	Description	Default
item	Union[bytes, Iterator[bytes]]	bytes object or iterator yielding bytes.	required

Source code in bamboo/io.py

def append(self, item: t.Union[bytes, t.Iterator[bytes]]) -> None:
    """Add bytes or iterator yielding bytes inside.

    Args:
        item: bytes object or iterator yielding bytes.
    """
    if isinstance(item, bytes):
        self._iters.append(
            BufferedBinaryIterator(item, bufsize=self._bufsize)
        )
    else:
        self._iters.append(
            BufferedIteratorWrapper(item, bufsize=self._bufsize)
        )

BufferedFileIterator

Iterator of binary made with file path.

Its object can be used to iterate file-like obj made with file path.

__init__(self, path, bufsize=8192, remove=False) special

Parameters:

Name	Type	Description	Default
path	str	File path.	required
bufsize	int	Chunk size of bytes objects yielded from the iterator.	8192

Source code in bamboo/io.py

def __init__(
    self,
    path: str,
    bufsize: int = 8192,
    remove: bool = False,
) -> None:
    """
    Args:
        path: File path.
        bufsize: Chunk size of bytes objects yielded from the iterator.
    """
    self._path = path
    self._remove = remove

    file = open(path, "br")
    super().__init__(file, bufsize=bufsize)

BufferedIteratorWrapper

Iterator wrapping iterator yeilding bytes objects.

Its object holds an iterator inner, which can yeilds bytes objects with arbitary sizes, and its object yeilds bytes objects with specified chunk size, i.e. this object changes chunk size of bytes yielded from given iterator.

__init__(self, iter, bufsize=8192) special

Parameters:

Name	Type	Description	Default
iter	Iterator[bytes]	Iterator wrapped.	required
bufsize	int	Chunk size of bytes objects yielded from the iterator.	8192

Source code in bamboo/io.py

def __init__(self, iter: t.Iterator[bytes], bufsize: int = 8192) -> None:
    """
    Args:
        iter: Iterator wrapped.
        bufsize: Chunk size of bytes objects yielded from the iterator.
    """
    super().__init__()

    self._iter = iter
    self._bufsize = bufsize
    self._buffer = io.BytesIO()

BufferedStreamIterator

Iterator of binary made with file-like object.

Its object can be used to iterate file-like obj of binary.

__init__(self, stream, bufsize=8192) special

Parameters:

Name	Type	Description	Default
stream	~BinaryReadableStream_t	File-like object readable binary.	required
bufsize	int	Chunk size of bytes objects yielded from the iterator.	8192

Exceptions:

Type	Description
TypeError	Raised if stream is not readable.

Source code in bamboo/io.py

def __init__(
    self,
    stream: BinaryReadableStream_t,
    bufsize: int = 8192
) -> None:
    """
    Args:
        stream: File-like object readable binary.
        bufsize: Chunk size of bytes objects yielded from the iterator.

    Raises:
        TypeError: Raised if `stream` is not readable.
    """
    super().__init__()

    if not (isinstance(stream, io.IOBase) and hasattr(stream, "read")):
        raise TypeError(
            "'stream' must be IOBase and have the 'read' method."
        )

    self._stream = stream
    self._bufsize = bufsize

ResponseBodyIteratorBase

Base class of iterator for buffering response body.