File support for asyncio applications providing async/await compatible file operations.
npx @tessl/cli install tessl/pypi-aiofiles@24.1.0File support for asyncio applications providing async/await compatible file operations. aiofiles wraps standard Python file I/O operations and delegates them to a thread pool executor, preventing blocking of the asyncio event loop during file operations.
pip install aiofilesimport aiofilesFor temporary file operations:
import aiofiles.tempfileFor OS file system operations:
import aiofiles.os
from aiofiles import ospathimport asyncio
import aiofiles
async def main():
# Basic file reading
async with aiofiles.open('example.txt', mode='r') as f:
contents = await f.read()
print(contents)
# Basic file writing
async with aiofiles.open('output.txt', mode='w') as f:
await f.write('Hello, async world!')
# Line-by-line reading with async iteration
async with aiofiles.open('data.txt', mode='r') as f:
async for line in f:
print(line.strip())
asyncio.run(main())aiofiles uses a delegation pattern where all I/O operations are wrapped and executed in a thread pool executor:
This design ensures that file operations don't block the asyncio event loop while maintaining an API nearly identical to Python's built-in file operations.
Basic file opening, reading, writing, and access to standard streams. These operations form the foundation of aiofiles functionality.
def open(
file,
mode: str = "r",
buffering: int = -1,
encoding: str = None,
errors: str = None,
newline: str = None,
closefd: bool = True,
opener = None,
*,
loop = None,
executor = None
) -> AiofilesContextManager: ...
# Standard stream objects
stdin: AsyncTextIndirectIOWrapper
stdout: AsyncTextIndirectIOWrapper
stderr: AsyncTextIndirectIOWrapper
stdin_bytes: AsyncIndirectBufferedIOBase
stdout_bytes: AsyncIndirectBufferedIOBase
stderr_bytes: AsyncIndirectBufferedIOBaseAsync interface to Python's tempfile module, providing temporary files and directories with automatic cleanup.
def NamedTemporaryFile(
mode: str = "w+b",
buffering: int = -1,
encoding: str = None,
newline: str = None,
suffix: str = None,
prefix: str = None,
dir: str = None,
delete: bool = True,
delete_on_close: bool = True, # Python 3.12+ only
*,
loop = None,
executor = None
) -> AiofilesContextManager: ...
def TemporaryFile(
mode: str = "w+b",
buffering: int = -1,
encoding: str = None,
newline: str = None,
suffix: str = None,
prefix: str = None,
dir: str = None,
*,
loop = None,
executor = None
) -> AiofilesContextManager: ...Async versions of os module functions for file system operations like renaming, removing, creating directories, and getting file statistics.
async def stat(path: str, *, loop = None, executor = None): ...
async def rename(src: str, dst: str, *, loop = None, executor = None): ...
async def remove(path: str, *, loop = None, executor = None): ...
async def mkdir(path: str, mode: int = 0o777, *, loop = None, executor = None): ...
async def listdir(path: str = ".", *, loop = None, executor = None): ...Async versions of os.path module functions for path manipulation and information gathering.
async def exists(path: str, *, loop = None, executor = None) -> bool: ...
async def isfile(path: str, *, loop = None, executor = None) -> bool: ...
async def isdir(path: str, *, loop = None, executor = None) -> bool: ...
async def getsize(path: str, *, loop = None, executor = None) -> int: ...
async def getmtime(path: str, *, loop = None, executor = None) -> float: ...class AiofilesContextManager:
"""Async context manager for aiofiles operations."""
def __init__(self, coro): ...
async def __aenter__(self): ...
async def __aexit__(self, exc_type, exc_val, exc_tb): ...
def __await__(self): ...
class AsyncBase:
"""Base class for async file wrappers."""
def __init__(self, file, loop, executor): ...
def __aiter__(self): ...
async def __anext__(self): ...