aenumerate - enumerate for async for

aenumerate.py

"""aenumerate - enumerate for async for"""

import asyncio
from collections.abc import AsyncIterable, AsyncIterator

# Thanks to @artpods56 for this version
async def aenumerate[T](iterable: AsyncIterable[T], start: int = 0) -> AsyncIterator[tuple[int, T]]:
    """Asynchronously enumerate an async iterator from a given start value"""
    i = start
    async for item in iterable:
        yield i, item
        i += 1

# Example usage
async def iter_lines(host, port):
    """Iterator over lines from host:port, print them with line number"""
    rdr, wtr = await asyncio.open_connection(host, port)
    async for lnum, line in aenumerate(rdr, 1):
        line = line.decode().rstrip()
        print('[{}:{}] {:02d} {}'.format(host, port, lnum, line))


if __name__ == '__main__':
    from argparse import ArgumentParser
    parser = ArgumentParser(description='enumerate lines from TCP server')
    parser.add_argument('host', help='host to connect to')
    parser.add_argument('port', help='port to connect to', type=int)
    args = parser.parse_args()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(iter_lines(args.host, args.port))
    loop.close()

# Run server: nc -lc -p 7654 < some-file
# (or on osx: nc -l 7654 < some-file)

With typehint:

_T = typing.TypeVar("_T")

async def aenumerate(iterable: collections.abc.AsyncIterator[_T], start=0):
    """Asynchronously enumerate an async iterator from a given start value"""
    i = start
    async for item in iterable:
        yield i, item
        i += 1

_T = typing.TypeVar("_T")

async def aenumerate(iterable: collections.abc.AsyncIterable[_T], start: int = 0) -> collections.abc.AsyncIterator[Tuple[int, _T]]:
    """Asynchronously enumerate an async iterator from a given start value"""
    i = start
    async for item in iterable:
        yield i, item
        i += 1

cleaned up

from collections.abc import AsyncIterable, AsyncIterator

async def aenumerate[T](iterable: AsyncIterable[T], start: int = 0) -> AsyncIterator[tuple[int, T]]:
    """Asynchronously enumerate an async iterator from a given start value"""
    i = start
    async for item in iterable:
        yield i, item
        i += 1

Thanks, updated.