The goal of the code is to read a text file asynchronously, allowing lines to be processed while other lines are loaded in the background, with support for buffering, encoding, and detailed logging settings.

async_read_txt_file.py

import logging
import sys
import asyncio
import time
from contextlib import asynccontextmanager

logging.basicConfig(level=logging.DEBUG, stream=sys.stdout,
                    format='[%(asctime)s]%(levelname)s(%(name)s): %(message)s')
LOGGER = logging.getLogger(__name__)

@asynccontextmanager
async def async_read_txt_file(filename: str,
                              buffer_hint: int = -1,
                              encoding='utf-8',
                              errors=None,
                              verbose=False):
    kwargs = {'mode': 'rt'}
    if encoding is not None:
        kwargs.update({'encoding': encoding})
    if errors is not None:
        kwargs.update({'errors': errors})
    with open(filename, **kwargs) as opened_file:
        def _readlines_():
            if verbose:
                LOGGER.info(f"Reading lines from file {filename}")
            # may be slow as it has disk access
            lines = opened_file.readlines(buffer_hint)
            for i in range(15):
                LOGGER.info("Simulating reading a file with sleep(1)...")
                time.sleep(1)
            if verbose:
                if lines:
                    LOGGER.info(f"{len(lines)} lines read from file {filename}")
                else:
                    LOGGER.info(f"End of file reading: {filename}")
            return lines
        async def _gen_():
            import gc
            lines = _readlines_()
            task = None
            while lines:
                task = asyncio.gather(asyncio.to_thread(_readlines_))
                for line in lines:
                    yield line
                lines = await task
                lines = lines[0]
            del task
            del lines
            gc.collect()
        yield _gen_()


async def main():
    async with async_read_txt_file("main.py",
                                   buffer_hint=256,
                                   verbose=True) as opened_file:
        async for line in opened_file:
            # faster process on read text result
            print(line.rstrip())
            await asyncio.sleep(1)
        print("END!")
    
if __name__ == "__main__":
    asyncio.run(main())