File support for asyncio applications providing async/await compatible file operations.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Basic file opening, reading, writing, and access to standard streams. These operations form the foundation of aiofiles functionality and provide async versions of Python's built-in file operations.
Opens files asynchronously with the same signature as Python's built-in open() function, but returns an async context manager.
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:
"""
Open file and return async context manager.
Parameters:
- file: File path or file descriptor to open
- mode: File open mode ('r', 'w', 'a', 'rb', 'wb', etc.)
- buffering: Buffer size (-1 for default, 0 for unbuffered, >0 for buffer size)
- encoding: Text encoding (for text mode files)
- errors: How to handle encoding errors ('strict', 'ignore', 'replace')
- newline: How to handle newlines (None, '', '\n', '\r', '\r\n')
- closefd: Whether to close file descriptor when file is closed
- opener: Custom opener function
- loop: Event loop to use (defaults to current running loop)
- executor: Thread pool executor to use (defaults to loop's default executor)
Returns:
AiofilesContextManager that yields async file object when awaited
"""Usage Example:
import aiofiles
# Text file operations
async with aiofiles.open('data.txt', mode='r') as f:
content = await f.read()
async with aiofiles.open('output.txt', mode='w', encoding='utf-8') as f:
await f.write('Hello, world!')
# Binary file operations
async with aiofiles.open('image.jpg', mode='rb') as f:
data = await f.read()
async with aiofiles.open('data.bin', mode='wb') as f:
await f.write(b'Binary data')Async access to standard input, output, and error streams including both text and binary interfaces.
stdin: AsyncTextIndirectIOWrapper
"""Async text interface to sys.stdin"""
stdout: AsyncTextIndirectIOWrapper
"""Async text interface to sys.stdout"""
stderr: AsyncTextIndirectIOWrapper
"""Async text interface to sys.stderr"""
stdin_bytes: AsyncIndirectBufferedIOBase
"""Async binary interface to sys.stdin.buffer"""
stdout_bytes: AsyncIndirectBufferedIOBase
"""Async binary interface to sys.stdout.buffer"""
stderr_bytes: AsyncIndirectBufferedIOBase
"""Async binary interface to sys.stderr.buffer"""Usage Example:
import aiofiles
# Reading from stdin
line = await aiofiles.stdin.readline()
print(f"You entered: {line.strip()}")
# Writing to stdout
await aiofiles.stdout.write("Hello from async stdout!\n")
await aiofiles.stdout.flush()
# Writing to stderr
await aiofiles.stderr.write("Error message\n")
# Binary operations
binary_data = await aiofiles.stdin_bytes.read(1024)
await aiofiles.stdout_bytes.write(b"Binary output\n")All async file objects returned by aiofiles.open() provide the following interface:
async def read(size: int = -1):
"""Read up to size bytes. If size is -1, read entire file."""
async def read1(size: int = -1):
"""Read up to size bytes using single system call (binary files only)."""
async def readall():
"""Read entire file (binary files only)."""
async def readinto(b):
"""Read into existing buffer (binary files only)."""
async def readline(size: int = -1):
"""Read one line up to size characters."""
async def readlines(hint: int = -1):
"""Read lines into list, with optional size hint."""async def write(data):
"""Write data to file. Returns number of characters/bytes written."""
async def writelines(lines):
"""Write list of lines to file."""
async def flush():
"""Flush write buffers to disk."""async def seek(offset: int, whence: int = 0):
"""Seek to position. whence: 0=start, 1=current, 2=end."""
async def tell() -> int:
"""Get current file position."""
async def truncate(size: int = None):
"""Truncate file to size bytes (current position if None)."""
async def close():
"""Close the file."""async def seekable() -> bool:
"""Return whether file supports seeking."""
async def readable() -> bool:
"""Return whether file supports reading."""
async def writable() -> bool:
"""Return whether file supports writing."""
async def isatty() -> bool:
"""Return whether file is a TTY device."""
def fileno() -> int:
"""Return file descriptor number."""
# Properties (synchronous access)
closed: bool # Whether file is closed
name: str # File name
mode: str # File mode
encoding: str # Text encoding (text files only)
errors: str # Error handling mode (text files only)
newlines: str # Newline handling (text files only)All async file objects support async iteration for line-by-line reading:
async with aiofiles.open('data.txt') as f:
async for line in f:
print(line.strip())File operations may raise standard Python I/O exceptions:
FileNotFoundError: File does not existPermissionError: Insufficient permissionsIsADirectoryError: Expected file but path is directoryOSError: General OS-level errorsUnicodeError: Text encoding/decoding errorsValueError: Invalid arguments (e.g., invalid mode)Usage Example:
import aiofiles
try:
async with aiofiles.open('nonexistent.txt', 'r') as f:
content = await f.read()
except FileNotFoundError:
print("File not found")
except PermissionError:
print("Permission denied")
except OSError as e:
print(f"OS error: {e}")Install with Tessl CLI
npx tessl i tessl/pypi-aiofiles