tessl/pypi-aiofile
Asynchronous file operations with asyncio support.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
high-level-interface.mddocs/
High-Level File Interface
Python file-like interface with asynchronous methods, providing familiar patterns for developers while maintaining async/await compatibility. The high-level interface mimics Python's built-in file objects but with async methods.
Capabilities
async_open Function
High-level helper that mimics Python's built-in open() function but returns file-like objects with async methods.
def async_open(
file_specifier: Union[str, Path, FileIOType],
mode: str = "r",
*args,
**kwargs
) -> Union[BinaryFileWrapper, TextFileWrapper]:
"""
Open file and return appropriate wrapper with async methods.
Args:
file_specifier: File path or existing file object
mode: File mode string (same as built-in open())
*args: Additional arguments passed to AIOFile constructor
**kwargs: Additional keyword arguments passed to AIOFile constructor
Returns:
BinaryFileWrapper for binary modes ('rb', 'wb', etc.)
TextFileWrapper for text modes ('r', 'w', etc.)
Raises:
ValueError: If args provided when wrapping existing file object
"""Base File Wrapper
Abstract base class providing common functionality for both binary and text file wrappers.
class FileIOWrapperBase:
def __init__(self, afp: AIOFile, *, offset: int = 0):
"""
Initialize file wrapper.
Args:
afp: AIOFile instance to wrap
offset: Initial file position (0 for append mode sets to file end)
"""
def seek(self, offset: int) -> None:
"""
Set file position.
Args:
offset: New position in bytes
"""
def tell(self) -> int:
"""
Get current file position.
Returns:
Current position in bytes
"""
async def flush(self, sync_metadata: bool = False) -> None:
"""
Flush file buffers to disk.
Args:
sync_metadata: If True, sync metadata (slower but more complete)
"""
async def close(self) -> None:
"""Close the underlying file."""
def iter_chunked(self, chunk_size: int = 32768) -> Reader:
"""
Create chunked reader from current position.
Args:
chunk_size: Size of each chunk in bytes
Returns:
Reader instance for chunked iteration
"""Binary File Wrapper
File-like interface for binary files with async methods.
class BinaryFileWrapper(FileIOWrapperBase):
def __init__(self, afp: AIOFile):
"""
Initialize binary file wrapper.
Args:
afp: AIOFile instance in binary mode
Raises:
ValueError: If AIOFile is not in binary mode
"""
async def read(self, length: int = -1) -> bytes:
"""
Read binary data from file.
Args:
length: Number of bytes to read (-1 for all remaining)
Returns:
Binary data as bytes
"""
async def write(self, data: bytes) -> int:
"""
Write binary data to file.
Args:
data: Binary data to write
Returns:
Number of bytes written
"""
async def readline(self, size: int = -1, newline: bytes = b"\n") -> bytes:
"""
Read one line from binary file.
Args:
size: Maximum bytes to read (-1 for no limit)
newline: Line separator bytes
Returns:
Line as bytes including separator
"""Text File Wrapper
File-like interface for text files with async methods and proper encoding handling.
class TextFileWrapper(FileIOWrapperBase):
def __init__(self, afp: AIOFile):
"""
Initialize text file wrapper.
Args:
afp: AIOFile instance in text mode
Raises:
ValueError: If AIOFile is in binary mode
"""
@property
def encoding(self) -> str:
"""Text encoding of the file."""
async def read(self, length: int = -1) -> str:
"""
Read text data from file.
Args:
length: Number of characters to read (-1 for all remaining)
Returns:
Text data as string
"""
async def write(self, data: str) -> int:
"""
Write text data to file.
Args:
data: Text data to write
Returns:
Number of bytes written (encoded size)
"""
async def readline(self, size: int = -1, newline: str = "\n") -> str:
"""
Read one line from text file.
Args:
size: Maximum characters to read (-1 for no limit)
newline: Line separator string
Returns:
Line as string including separator
"""Usage Examples
Basic Text File Operations
import asyncio
from aiofile import async_open
async def text_operations():
# Read entire text file
async with async_open('document.txt', 'r') as file:
content = await file.read()
print(content)
# Write text file
async with async_open('output.txt', 'w') as file:
await file.write("Hello, async world!\n")
await file.write("Second line\n")
await file.flush() # Ensure data is written
asyncio.run(text_operations())Binary File Operations
import asyncio
from aiofile import async_open
async def binary_operations():
# Read binary file
async with async_open('image.png', 'rb') as file:
header = await file.read(8) # Read PNG header
print(f"Header: {header.hex()}")
# Write binary file
async with async_open('output.bin', 'wb') as file:
data = bytes(range(256)) # Create sample binary data
await file.write(data)
asyncio.run(binary_operations())Line-by-Line Processing
import asyncio
from aiofile import async_open
async def process_lines():
async with async_open('large_file.txt', 'r') as file:
# Method 1: Using async iteration (recommended)
async for line in file:
print(f"Line: {line.strip()}")
# Method 2: Manual readline
file.seek(0) # Reset to beginning
while True:
line = await file.readline()
if not line:
break
print(f"Manual: {line.strip()}")
asyncio.run(process_lines())File Position Management
import asyncio
from aiofile import async_open
async def position_example():
async with async_open('data.txt', 'r+') as file:
# Read from beginning
data1 = await file.read(10)
print(f"Position after read: {file.tell()}")
# Seek to specific position
file.seek(5)
data2 = await file.read(5)
print(f"Data from position 5: {data2}")
# Write at current position
await file.write(" [INSERTED] ")
print(f"Position after write: {file.tell()}")
asyncio.run(position_example())Chunked Reading
import asyncio
from aiofile import async_open
async def chunked_reading():
async with async_open('large_file.txt', 'r') as file:
# Create chunked reader
reader = file.iter_chunked(chunk_size=1024)
# Process chunks
async for chunk in reader:
print(f"Chunk size: {len(chunk)}")
# Process chunk...
asyncio.run(chunked_reading())Append Mode Operations
import asyncio
from aiofile import async_open
async def append_operations():
# Append to existing file
async with async_open('log.txt', 'a') as file:
print(f"Starting position: {file.tell()}") # At end of file
await file.write("New log entry\n")
await file.flush()
asyncio.run(append_operations())Working with Existing File Objects
import asyncio
from aiofile import async_open
async def wrap_existing_file():
# Wrap existing file object
with open('temp.txt', 'w+') as fp:
fp.write("Initial content")
fp.seek(0)
# Wrap with async interface
async_file = async_open(fp)
content = await async_file.read()
print(f"Content: {content}")
asyncio.run(wrap_existing_file())Error Handling
import asyncio
from aiofile import async_open
async def error_handling():
try:
async with async_open('nonexistent.txt', 'r') as file:
content = await file.read()
except FileNotFoundError:
print("File not found")
try:
async with async_open('readonly.txt', 'w') as file:
await file.write("data")
except PermissionError:
print("Permission denied")
asyncio.run(error_handling())Context Manager Support
All file wrapper classes support async context managers for automatic resource management:
# Automatic file opening and closing
async with async_open('file.txt', 'r') as file:
# File is automatically opened
data = await file.read()
# File is automatically closed and flushed when exiting contextThe context manager ensures:
- File is opened when entering the context
- File is properly closed when exiting the context
- Data is flushed for writable files before closing
- Resources are cleaned up even if exceptions occur
Install with Tessl CLI
npx tessl i tessl/pypi-aiofile