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

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

AIOFile

Name: tessl/pypi-aiofile
Author: tessl

Real asynchronous file operations with asyncio support. AIOFile provides comprehensive asynchronous file I/O capabilities with multiple backends for optimal performance across different platforms, featuring both low-level operations with explicit offset/chunk control and high-level file-like interfaces.

Package Information

Package Name: aiofile
Language: Python
Installation: pip install aiofile
Dependencies: caio (~0.9.0)

Core Imports

import aiofile

Common imports for different usage patterns:

# High-level file-like interface
from aiofile import async_open

# Low-level AIOFile class
from aiofile import AIOFile

# Streaming operations
from aiofile import Reader, Writer, LineReader

# File wrappers
from aiofile import BinaryFileWrapper, TextFileWrapper

Basic Usage

High-Level Usage (Recommended)

import asyncio
from aiofile import async_open

async def main():
    # Read a text file
    async with async_open('data.txt', 'r') as file:
        content = await file.read()
        print(content)
    
    # Write to a binary file
    async with async_open('output.bin', 'wb') as file:
        await file.write(b'Hello, async world!')
        
    # Read file line by line
    async with async_open('large_file.txt', 'r') as file:
        async for line in file:
            print(line.strip())

asyncio.run(main())

Low-Level Usage

import asyncio
from aiofile import AIOFile

async def main():
    # Low-level operations with explicit offset
    async with AIOFile('data.txt', 'r') as aio_file:
        # Read 100 bytes from offset 0
        data = await aio_file.read(100, 0)
        
        # Read another chunk from offset 100
        more_data = await aio_file.read(100, 100)

asyncio.run(main())

Architecture

AIOFile utilizes multiple backends for optimal performance across platforms:

Linux: Uses libaio (Linux AIO) for maximum performance with kernel-level asynchronous operations
POSIX (macOS): Uses threadpool-based implementation for compatibility
Fallback: Pure Python thread-based implementation for other platforms
Automatic Selection: Backend is chosen automatically based on system compatibility

The library provides three levels of abstraction:

Low-Level: Direct AIOFile operations with explicit offset and chunk size control
High-Level: File-like wrappers (BinaryFileWrapper, TextFileWrapper) accessible via async_open
Streaming: Async iterators (Reader, Writer, LineReader) for chunked operations

Capabilities

Core Low-Level File Operations

Direct asynchronous file operations with explicit offset and chunk size control. Provides maximum performance and flexibility for advanced use cases requiring precise control over file I/O operations.

class AIOFile:
    def __init__(
        self, 
        filename: Union[str, Path], 
        mode: str = "r", 
        encoding: str = "utf-8",
        context: Optional[AsyncioContextBase] = None,
        executor: Optional[Executor] = None
    ): ...
    
    async def open(self) -> Optional[int]: ...
    async def close(self) -> None: ...
    async def read(self, size: int = -1, offset: int = 0) -> Union[bytes, str]: ...
    async def write(self, data: Union[str, bytes], offset: int = 0) -> int: ...
    async def fsync(self) -> None: ...
    async def fdsync(self) -> None: ...

Core Operations

High-Level File Interface

Python file-like interface with asynchronous methods, providing familiar patterns for developers while maintaining async/await compatibility. Includes both text and binary file support with proper encoding handling.

def async_open(
    file_specifier: Union[str, Path, FileIOType],
    mode: str = "r", 
    *args, 
    **kwargs
) -> Union[BinaryFileWrapper, TextFileWrapper]: ...

class BinaryFileWrapper:
    async def read(self, length: int = -1) -> bytes: ...
    async def write(self, data: bytes) -> int: ...
    async def readline(self, size: int = -1, newline: bytes = b"\n") -> bytes: ...

class TextFileWrapper:
    async def read(self, length: int = -1) -> str: ...
    async def write(self, data: str) -> int: ...
    async def readline(self, size: int = -1, newline: str = "\n") -> str: ...

High-Level Interface

Streaming Operations

Async iterators for chunk-based reading and sequential writing operations. Provides efficient memory usage for large files and streaming data processing with configurable chunk sizes.

class Reader:
    def __init__(
        self, 
        aio_file: AIOFile, 
        offset: int = 0, 
        chunk_size: int = 32768
    ): ...
    async def read_chunk(self) -> Union[str, bytes]: ...

class Writer:
    def __init__(self, aio_file: AIOFile, offset: int = 0): ...
    async def __call__(self, data: Union[str, bytes]) -> None: ...

class LineReader:
    def __init__(
        self, 
        aio_file: AIOFile, 
        offset: int = 0, 
        chunk_size: int = 4192, 
        line_sep: str = "\n"
    ): ...
    async def readline(self) -> Union[str, bytes]: ...

Streaming Operations

Types

FileIOType = Union[TextIO, BinaryIO]

class FileMode:
    readable: bool
    writable: bool
    plus: bool
    appending: bool
    created: bool
    flags: int
    binary: bool

def parse_mode(mode: str) -> FileMode: ...

Version Information

__version__: str
__author__: str
version_info: Tuple[int, ...]
author_info: List[Tuple[str, str]]
package_info: str
package_license: str
project_home: str
team_email: str

Error Handling

AIOFile properly handles and propagates file system errors:

OSError: For system-level file operation errors
ValueError: For invalid parameters (negative size, wrong data types for binary/text mode)
UnicodeDecodeError: For encoding issues in text mode (automatically retried with larger chunks)
asyncio.InvalidStateError: For operations on closed files

Context Management

All file objects support async context managers for automatic resource cleanup:

async with AIOFile('file.txt', 'r') as aio_file:
    # File is automatically opened
    data = await aio_file.read()
# File is automatically closed and synced