zreader

A flexible, type-safe Python library for reading data from various sources, supporting both synchronous and asynchronous environments.

Features

• Unified Reader Interface: Define your own readers by subclassing BaseReader.
• Sync and Async Support: Use .read() for synchronous iteration, or .aread() for asynchronous iteration.
• Filtering and Limiting: Both methods support filter_fn and limit arguments.
• Type Safety: Uses Python generics and protocols for type checking.

Quickstart

Synchronous Usage

>>> from zreader.readers.list_reader import ListReader
>>> data = ["apple", "banana", "apricot", "blueberry"]
>>> reader = ListReader(data)
>>> for item in reader.read(filter_fn=lambda x: x.startswith("a")):
...     print(item)
apple
apricot
>>> for item in reader.read(limit=2):
...     print(item)
apple
banana
>>> list(reader.read(filter_fn=lambda x: x.startswith("a")))
['apple', 'apricot']
>>> list(reader.read(limit=2))
['apple', 'banana']

Asynchronous Usage

>>> import asyncio
>>> from zreader.readers.list_reader import ListReader
>>> async def main():
...     data = ["apple", "banana", "apricot", "blueberry"]
...     reader = ListReader(data)
...     async for item in reader.aread(filter_fn=lambda x: x.startswith("a")):
...         print(item)
...     items = [item async for item in reader.aread(filter_fn=lambda x: x.startswith("a"))]
...     print(items)
>>> asyncio.run(main())
apple
apricot
['apple', 'apricot']

Creating Your Own Reader

Subclass BaseReader and implement _read (sync) and/or _aread (async). Here's an example of an async reader that fetches posts from JSONPlaceholder:

>>> import aiohttp
>>> from zreader.base import BaseReader
>>> from typing import AsyncIterator
>>> class JSONPlaceholderPostsReader(BaseReader[dict]):
...     def _read(self):
...         raise NotImplementedError("Sync read not implemented for this reader.")
...     async def _aread(self) -> AsyncIterator[dict]:
...         url = f"https://jsonplaceholder.typicode.com/posts"
...         async with aiohttp.ClientSession() as session:
...             async with session.get(url) as resp:
...                 posts = await resp.json()
...                 for post in posts:
...                     yield post
>>> import asyncio
>>> async def main():
...     reader = JSONPlaceholderPostsReader()
...     results = [post async for post in reader.aread(limit=5)]
...     print(len(results))
>>> asyncio.run(main())
5

This example uses aiohttp for async HTTP requests. You can adapt this pattern for any async data source.

API Reference

BaseReader

• read(limit=None, filter_fn=None) -> Iterator[T]: Synchronous iterator.
• aread(limit=None, filter_fn=None) -> AsyncIterator[T]: Asynchronous iterator.

ListReader

A built-in reader for lists of strings. See examples above.

---

For more, see the source code and tests.