Core Low-Level File 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
    ):
        """
        Initialize AIOFile instance.
        
        Args:
            filename: Path to file as string or Path object
            mode: File mode ('r', 'w', 'a', 'x', 'b', '+' combinations)
            encoding: Text encoding for non-binary modes (default: 'utf-8')
            context: Optional caio AsyncioContext for custom backend configuration
            executor: Optional executor for thread-based operations
        """

async def open(self) -> Optional[int]:
    """
    Open the file for operations.
    
    Returns:
        File descriptor number if newly opened, None if already open
        
    Raises:
        asyncio.InvalidStateError: If file was previously closed
    """

async def close(self) -> None:
    """
    Close the file and sync data if writable.
    
    Automatically calls fdsync() for writable files before closing.
    """

def fileno(self) -> int:
    """
    Get the file descriptor number.
    
    Returns:
        File descriptor as integer
        
    Raises:
        asyncio.InvalidStateError: If file is not open
    """

async def read(self, size: int = -1, offset: int = 0) -> Union[bytes, str]:
    """
    Read data from file at specified offset.
    
    Args:
        size: Number of bytes to read (-1 for entire file)
        offset: Byte offset to start reading from
    
    Returns:
        bytes in binary mode, str in text mode
        
    Raises:
        ValueError: If size < -1
    """

async def read_bytes(self, size: int = -1, offset: int = 0) -> bytes:
    """
    Read raw bytes from file at specified offset.
    
    Args:
        size: Number of bytes to read (-1 for entire file)
        offset: Byte offset to start reading from
    
    Returns:
        Raw bytes data
        
    Raises:
        ValueError: If size < -1
    """

async def write(self, data: Union[str, bytes], offset: int = 0) -> int:
    """
    Write data to file at specified offset.
    
    Args:
        data: Data to write (str for text mode, bytes for binary mode)
        offset: Byte offset to start writing at
    
    Returns:
        Number of bytes written
        
    Raises:
        ValueError: If data type doesn't match file mode
    """

async def write_bytes(self, data: bytes, offset: int = 0) -> int:
    """
    Write raw bytes to file at specified offset.
    
    Handles partial writes by continuing until all data is written.
    
    Args:
        data: Raw bytes to write
        offset: Byte offset to start writing at
    
    Returns:
        Number of bytes written
        
    Raises:
        RuntimeError: If write operation returns 0 (disk full, etc.)
        OSError: For system-level write errors
    """

async def fsync(self) -> None:
    """
    Sync file data and metadata to disk.
    
    Forces all written data and metadata to be physically stored.
    More comprehensive than fdsync() but potentially slower.
    """

async def fdsync(self) -> None:
    """
    Sync file data to disk (without metadata).
    
    Forces written data to be physically stored.
    Faster than fsync() but doesn't guarantee metadata sync.
    """

async def truncate(self, length: int = 0) -> None:
    """
    Truncate file to specified length.
    
    Args:
        length: New file length in bytes (default: 0)
    """

def encode_bytes(self, data: str) -> bytes:
    """
    Encode string to bytes using file's encoding.
    
    Args:
        data: String to encode
    
    Returns:
        Encoded bytes
    """

def decode_bytes(self, data: bytes) -> str:
    """
    Decode bytes to string using file's encoding.
    
    Args:
        data: Bytes to decode
    
    Returns:
        Decoded string
    """

@classmethod
def from_fp(cls, fp: FileIOType, **kwargs) -> "AIOFile":
    """
    Create AIOFile from existing file pointer.
    
    Args:
        fp: Existing file object (TextIO or BinaryIO)
        **kwargs: Additional arguments passed to constructor
    
    Returns:
        AIOFile instance wrapping the file pointer
    """

@property
def name(self) -> str:
    """File name as string."""

@property
def loop(self) -> asyncio.AbstractEventLoop:
    """Associated event loop."""

@property
def encoding(self) -> str:
    """Text encoding for non-binary files."""

@property
def mode(self) -> FileMode:
    """Parsed file mode information."""

def create_context(
    max_requests: int = caio.AsyncioContext.MAX_REQUESTS_DEFAULT
) -> caio.AsyncioContext:
    """
    Create a new caio AsyncioContext.
    
    Args:
        max_requests: Maximum concurrent I/O requests
    
    Returns:
        New AsyncioContext instance
    """

def get_default_context() -> caio.AsyncioContext:
    """
    Get or create the default caio context for current event loop.
    
    Returns:
        Default AsyncioContext for the current event loop
    """

def parse_mode(mode: str) -> FileMode:
    """
    Parse file mode string into structured FileMode object.
    
    Args:
        mode: File mode string (e.g., 'r', 'wb', 'a+')
    
    Returns:
        FileMode namedtuple with parsed mode information
        
    Raises:
        Exception: For invalid mode combinations
    """

class FileMode:
    """
    Structured file mode information.
    """
    readable: bool      # File is readable
    writable: bool      # File is writable  
    plus: bool          # Mode includes '+'
    appending: bool     # File is in append mode
    created: bool       # File should be created (exclusive)
    flags: int          # OS-level file flags
    binary: bool        # File is in binary mode

import asyncio
from aiofile import AIOFile

async def basic_operations():
    # Open file for reading
    async with AIOFile('data.txt', 'r') as afile:
        # Read first 100 characters
        chunk1 = await afile.read(100, 0)
        
        # Read next 100 characters
        chunk2 = await afile.read(100, 100)
        
        # Read entire file from beginning
        entire_file = await afile.read(-1, 0)

asyncio.run(basic_operations())

import asyncio
from aiofile import AIOFile

async def write_operations():
    async with AIOFile('output.txt', 'w') as afile:
        # Write at beginning
        await afile.write("Hello, ", 0)
        
        # Write at offset 7
        await afile.write("World!", 7)
        
        # Force sync to disk
        await afile.fsync()

asyncio.run(write_operations())

import asyncio
from aiofile import AIOFile

async def binary_operations():
    async with AIOFile('data.bin', 'wb') as afile:
        # Write binary data
        data = b'\x00\x01\x02\x03\x04'
        bytes_written = await afile.write_bytes(data, 0)
        print(f"Wrote {bytes_written} bytes")
        
        # Sync data to disk
        await afile.fdsync()

asyncio.run(binary_operations())

AIO_FILE_NOT_OPENED = -1  # File state: not opened
AIO_FILE_CLOSED = -2      # File state: closed

FileIOType = Union[TextIO, BinaryIO]  # Type alias for file objects