Async wrappers for tempfile.NamedTemporaryFile and tempfile.NamedTemporaryFile.

async_temp.py

# Motivation: I had to use standard tarfile to create an archive within an async function.
# def pack(f):
#     with tarfile.open(output, "w:gz") as tar:
#         tar.add(src_dir, arcname="blah")
#
# async with temp_file() as f:
#     asyncio.get_running_loop().run_in_executor(None, pack, f)

@contextlib.asynccontextmanager
async def temp_file() -> Iterator[pathlib.Path]:
    """Yields a path of a temporary file.

    The file is removed upon the context manager exit.
    """
    # test exception is propagated
    loop, executor = asyncio.get_running_loop(), None
    file = await loop.run_in_executor(executor, tempfile.NamedTemporaryFile)
    try:
        await loop.run_in_executor(executor, file.__enter__)
        yield pathlib.Path(file.name)
    except Exception as err:
        await loop.run_in_executor(executor, file.__exit__, type(err), err, err.__traceback__)
        raise


@contextlib.asynccontextmanager
async def temp_directory() -> Iterator[pathlib.Path]:
    """Yields a path of a temporary directoy.

    The directory and all of its content is removed upon the context manager exit.
    """
    loop, executor = asyncio.get_running_loop(), None
    _dir = await loop.run_in_executor(executor, tempfile.NamedTemporaryFile)
    try:
        await loop.run_in_executor(executor, _dir.__enter__)
        yield pathlib.Path(_dir.name)
    except Exception as err:
        await loop.run_in_executor(executor, _dir.__exit__, type(err), err, err.__traceback__)
        raise