How to use the apraw.models.Streamable function in aPRAW

To help you get started, we’ve selected a few aPRAW examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Dan6erbond / aPRAW / tests / unit / models / helpers / test_streamable.py View on Github external
def __aiter__(self):
                    return self

                def __len__(self):
                    return len(self._items)

                async def __anext__(self):
                    if self._index >= len(self):
                        raise StopAsyncIteration

                    self._index += 1
                    return self._items[self._index - 1]

            return AIterator()

        streamable = Streamable(get_async_iterator)
        result = [i async for i in streamable()]
        assert result == items
github Dan6erbond / aPRAW / tests / unit / models / helpers / test_streamable.py View on Github external
async def test_streamable_sync_generator(self):
        items = list(range(20))

        def sync_generator():
            for i in items:
                yield i

        streamable = Streamable(sync_generator)
        result = [i async for i in streamable()]
        assert result == items
github Dan6erbond / aPRAW / tests / unit / models / helpers / test_streamable.py View on Github external
async def test_streamable_sync_list(self):
        items = list(range(20))

        def get_list():
            return items

        streamable = Streamable(get_list)
        result = [i async for i in streamable()]
        assert result == items
github Dan6erbond / aPRAW / tests / unit / models / helpers / test_streamable.py View on Github external
async def test_streamable_async_generator(self):
        items = list(range(20))

        async def async_generator():
            for i in items:
                yield i

        streamable = Streamable(async_generator)
        result = [i async for i in streamable()]
        assert result == items