advent22/api/asynccontextmanager.py

import asyncio
import functools
from io import BytesIO

from cache import AsyncLRU

FILES = __file__, "pyproject.toml"
REPS = 3

#
# sync impl
#


@functools.lru_cache
def get_bytes_sync(fn) -> bytes:
    print("sync open")
    with open(fn, "rb") as file:
        return file.readline()


def get_buf_sync(fn) -> BytesIO:
    return BytesIO(get_bytes_sync(fn))


def main_sync() -> None:
    for fn in FILES:
        for _ in range(REPS):
            print(get_buf_sync(fn).read())


#
# async impl
#


@AsyncLRU()
async def get_bytes_async(fn) -> bytes:
    print("async open")
    with open(fn, "rb") as file:
        return file.readline()


async def get_buf_async(fn) -> BytesIO:
    return BytesIO(await get_bytes_async(fn))


async def main_async() -> None:
    for fn in FILES:
        for _ in range(REPS):
            print((await get_buf_async(fn)).read())


if __name__ == "__main__":
    main_sync()
    asyncio.run(main_async())