Skip to content

Instantly share code, notes, and snippets.

@mezhaka
Created July 21, 2023 13:49
Show Gist options
  • Save mezhaka/963f81cdb06125d160b8a5844cb0b5fb to your computer and use it in GitHub Desktop.
Save mezhaka/963f81cdb06125d160b8a5844cb0b5fb to your computer and use it in GitHub Desktop.
Async wrappers for tempfile.NamedTemporaryFile and tempfile.NamedTemporaryFile.
# 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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment