aiopen

aiopen

aiopen ~ Async version of python's built in open -- based on aiofiles

Functions

aiopen.aiopen(file: PathType, mode: str = 'r', buffering: int = -1, encoding: Optional[str] = None, errors: None = None, newline: None = None, closefd: bool = True, opener: None = None, *, loop: Optional[AbstractEventLoop] = None, executor: Any = None) -> AiopenContextManager

Async version of the open builtin

Examples:

>>> async def main():
...     async with aiopen("test.txt", "w") as f:
...         await f.write("test")
...     async with aiopen("test.txt", "r") as f:
...         assert await f.read() == "test"
>>> asyncio.run(main())
>>> import os
>>> if os.path.exists("test.txt"):
...     os.remove("test.txt")

Source code in libs/aiopen/aiopen/core.py

def aiopen(
    file: PathType,
    mode: str = "r",
    buffering: int = -1,
    encoding: Optional[str] = None,
    errors: None = None,
    newline: None = None,
    closefd: bool = True,
    opener: None = None,
    *,
    loop: Optional[AbstractEventLoop] = None,
    executor: Any = None,
) -> AiopenContextManager:
    """Async version of the `open` builtin

    Examples:
        >>> async def main():
        ...     async with aiopen("test.txt", "w") as f:
        ...         await f.write("test")
        ...     async with aiopen("test.txt", "r") as f:
        ...         assert await f.read() == "test"
        >>> asyncio.run(main())
        >>> import os
        >>> if os.path.exists("test.txt"):
        ...     os.remove("test.txt")

    """
    return AiopenContextManager(
        _aiopen(
            str(file),
            mode=mode,
            buffering=buffering,
            encoding=encoding,
            errors=errors,
            newline=newline,
            closefd=closefd,
            opener=opener,
            loop=loop,
            executor=executor,
        )
    )

Modules

aiopen.__about__

Package metadata/info

aiopen.__main__

pkg entry ~ python -m aiopen

Functions

aiopen.__main__.main() -> None

Print package metadata

Source code in libs/aiopen/aiopen/__main__.py

def main() -> None:
    """Print package metadata"""
    import json

    sys.stdout.write(
        json.dumps(
            {
                "package": __title__,
                "version": __version__,
                "pkgroot": __pkgroot__,
            },
            indent=2,
        )
    )

aiopen.core

Async io base functions and utilities

Inspired by aiofiles

Classes

aiopen.core.AsyncBase(file: Union[BufferedWriter, TextIOWrapper, FileIO, BufferedRandom, BufferedReader], loop: AbstractEventLoop, executor: None)

Bases: Generic[AnyStr]

Source code in libs/aiopen/aiopen/core.py

def __init__(
    self,
    file: Union[
        BufferedWriter,
        TextIOWrapper,
        FileIO,
        BufferedRandom,
        BufferedReader,
    ],
    loop: AbstractEventLoop,
    executor: None,
) -> None:
    self._file = file
    self._loop = loop
    self._executor = executor

Functions

aiopen.core.AsyncBase.__anext__() -> AnyStr async

Simulate normal file iteration.

Source code in libs/aiopen/aiopen/core.py

async def __anext__(self) -> AnyStr:
    """Simulate normal file iteration."""
    line = await self.readline()
    if line:
        return cast(AnyStr, line)
    else:
        raise StopAsyncIteration

aiopen.core.BufferedIOAsyncBase(file: Union[BufferedWriter, TextIOWrapper, FileIO, BufferedRandom, BufferedReader], loop: AbstractEventLoop, executor: None)

Bases: AsyncBaseDetachable

Async version of io.BufferedWriter

Source code in libs/aiopen/aiopen/core.py

def __init__(
    self,
    file: Union[
        BufferedWriter,
        TextIOWrapper,
        FileIO,
        BufferedRandom,
        BufferedReader,
    ],
    loop: AbstractEventLoop,
    executor: None,
) -> None:
    self._file = file
    self._loop = loop
    self._executor = executor

Functions

aiopen.core.BufferedIOAsyncBase.__anext__() -> AnyStr async

Simulate normal file iteration.

Source code in libs/aiopen/aiopen/core.py

async def __anext__(self) -> AnyStr:
    """Simulate normal file iteration."""
    line = await self.readline()
    if line:
        return cast(AnyStr, line)
    else:
        raise StopAsyncIteration

aiopen.core.BufferedReaderAsync(file: Union[BufferedWriter, TextIOWrapper, FileIO, BufferedRandom, BufferedReader], loop: AbstractEventLoop, executor: None)

Bases: BufferedIOAsyncBase

The asyncio executor version of io.BufferedReader and Random.

Source code in libs/aiopen/aiopen/core.py

def __init__(
    self,
    file: Union[
        BufferedWriter,
        TextIOWrapper,
        FileIO,
        BufferedRandom,
        BufferedReader,
    ],
    loop: AbstractEventLoop,
    executor: None,
) -> None:
    self._file = file
    self._loop = loop
    self._executor = executor

Functions

aiopen.core.BufferedReaderAsync.__anext__() -> AnyStr async

Simulate normal file iteration.

Source code in libs/aiopen/aiopen/core.py

async def __anext__(self) -> AnyStr:
    """Simulate normal file iteration."""
    line = await self.readline()
    if line:
        return cast(AnyStr, line)
    else:
        raise StopAsyncIteration

aiopen.core.FileIOAsync(file: Union[BufferedWriter, TextIOWrapper, FileIO, BufferedRandom, BufferedReader], loop: AbstractEventLoop, executor: None)

Bases: AsyncBase

The asyncio executor version of io.FileIO.

Source code in libs/aiopen/aiopen/core.py

def __init__(
    self,
    file: Union[
        BufferedWriter,
        TextIOWrapper,
        FileIO,
        BufferedRandom,
        BufferedReader,
    ],
    loop: AbstractEventLoop,
    executor: None,
) -> None:
    self._file = file
    self._loop = loop
    self._executor = executor

Functions

aiopen.core.FileIOAsync.__anext__() -> AnyStr async

Simulate normal file iteration.

Source code in libs/aiopen/aiopen/core.py

async def __anext__(self) -> AnyStr:
    """Simulate normal file iteration."""
    line = await self.readline()
    if line:
        return cast(AnyStr, line)
    else:
        raise StopAsyncIteration

aiopen.core.TextIOWrapperAsync(file: Union[BufferedWriter, TextIOWrapper, FileIO, BufferedRandom, BufferedReader], loop: AbstractEventLoop, executor: None)

Bases: AsyncBase[str]

Async version of io.TextIOWrapper

Source code in libs/aiopen/aiopen/core.py

def __init__(
    self,
    file: Union[
        BufferedWriter,
        TextIOWrapper,
        FileIO,
        BufferedRandom,
        BufferedReader,
    ],
    loop: AbstractEventLoop,
    executor: None,
) -> None:
    self._file = file
    self._loop = loop
    self._executor = executor

Functions

aiopen.core.TextIOWrapperAsync.__anext__() -> AnyStr async

Simulate normal file iteration.

Source code in libs/aiopen/aiopen/core.py

async def __anext__(self) -> AnyStr:
    """Simulate normal file iteration."""
    line = await self.readline()
    if line:
        return cast(AnyStr, line)
    else:
        raise StopAsyncIteration

Functions

aiopen.core.aiopen(file: PathType, mode: str = 'r', buffering: int = -1, encoding: Optional[str] = None, errors: None = None, newline: None = None, closefd: bool = True, opener: None = None, *, loop: Optional[AbstractEventLoop] = None, executor: Any = None) -> AiopenContextManager

Async version of the open builtin

Examples:

>>> async def main():
...     async with aiopen("test.txt", "w") as f:
...         await f.write("test")
...     async with aiopen("test.txt", "r") as f:
...         assert await f.read() == "test"
>>> asyncio.run(main())
>>> import os
>>> if os.path.exists("test.txt"):
...     os.remove("test.txt")

Source code in libs/aiopen/aiopen/core.py

def aiopen(
    file: PathType,
    mode: str = "r",
    buffering: int = -1,
    encoding: Optional[str] = None,
    errors: None = None,
    newline: None = None,
    closefd: bool = True,
    opener: None = None,
    *,
    loop: Optional[AbstractEventLoop] = None,
    executor: Any = None,
) -> AiopenContextManager:
    """Async version of the `open` builtin

    Examples:
        >>> async def main():
        ...     async with aiopen("test.txt", "w") as f:
        ...         await f.write("test")
        ...     async with aiopen("test.txt", "r") as f:
        ...         assert await f.read() == "test"
        >>> asyncio.run(main())
        >>> import os
        >>> if os.path.exists("test.txt"):
        ...     os.remove("test.txt")

    """
    return AiopenContextManager(
        _aiopen(
            str(file),
            mode=mode,
            buffering=buffering,
            encoding=encoding,
            errors=errors,
            newline=newline,
            closefd=closefd,
            opener=opener,
            loop=loop,
            executor=executor,
        )
    )