Source code for trinity.buffer.buffer_reader

"""Reader of the buffer."""
from abc import ABC, abstractmethod
from typing import List, Optional

from trinity.common.constants import ReadStrategy



[docs]
class BufferReader(ABC):
    """Interface of the buffer reader."""


[docs]
    @abstractmethod
    def read(
        self, batch_size: Optional[int] = None, strategy: Optional[ReadStrategy] = None
    ) -> List:
        """Read from buffer."""



[docs]
    @abstractmethod
    async def read_async(
        self, batch_size: Optional[int] = None, strategy: Optional[ReadStrategy] = None
    ) -> List:
        """Read from buffer asynchronously."""