API Reference — Aquilia Documentation
Comprehensive guide and documentation for API Reference in the Aquilia framework. View API reference, examples, and implementation patterns.
Filesystem / API Reference Filesystem API Reference Detailed interface specifications for standalone async file operations, context-managed file handles, and directory path wrappers. Table of Contents , , , , , ].map((item, i) => ( • ))} Core Helper Functions High-level standalone functions available directly under aquilia.filesystem. from aquilia.filesystem import async_open, read_file, write_file async def async_open( path: str | Path, mode: str = "r", encoding: str | None = None, errors: str | None = None, buffering: int = -1, newline: str | None = None, ) -> AsyncFile: """Opens a file asynchronously. Yields an AsyncFile context manager.""" async def read_file( path: str | Path, encoding: str = "utf-8", errors: str = "strict", ) -> str | bytes: """Read and return entire file content. Returns bytes if binary mode is requested.""" async def write_file( path: str | Path, data: str | bytes, encoding: str = "utf-8", errors: str = "strict", atomic: bool = True, # Uses temp file + rename if True make_parents: bool = True, # Auto-creates parent directories ) -> None: """Writes content to a file.""" async def append_file( path: str | Path, data: str | bytes, encoding: str = "utf-8", errors: str = "strict", ) -> None: """Appends data to the end of a file.""" async def copy_file(src: str | Path, dst: str | Path) -> None: """Copy a file from source to destination.""" async def move_file(src: str | Path, dst: str | Path) -> None: """Move/rename a file atomically.""" async def delete_file(path: str | Path) -> None: """Delete a file. Idempotent: does NOT raise if file is missing.""" async def file_exists(path: str | Path) -> bool: """Returns True if the path exists and is a file.""" async def file_stat(path: str | Path) -> os.stat_result: """Retrieve file metadata (size, last modified time, etc.).""" FileSystem The DI-managed filesystem service class containing all query and operation methods. class FileSystem: def __init__(self, config: FileSystemConfig | None = None) -> None: """Create a new FileSystem service wrapper.""" async def open(self, path: str | Path, mode: str = "r", **kwargs: Any) -> AsyncFile: """Alias for async_open.""" async def read(self, path: str | Path, **kwargs: Any) -> str | bytes: """Alias for read_file.""" async def write(self, path: str | Path, data: str | bytes, **kwargs: Any) -> None: """Alias for write_file.""" async def list_dir(self, path: str | Path) -> list[str]: """List all filenames and subdirectories in a directory path.""" async def scan_dir(self, path: str | Path) -> AsyncIterator[DirEntry]: """Asynchronously iterate over entries (files/directories) in a path.""" async def make_dir(self, path: str | Path, parents: bool = True, exist_ok: bool = True) -> None: """Create a new directory.""" async def remove_dir(self, path: str | Path) -> None: """Remove an empty directory.""" async def remove_tree(self, path: str | Path) -> None: """Delete a directory and all of its contents recursively.""" async def copy_tree(self, src: str | Path, dst: str | Path) -> None: """Copy a directory tree recursively to another destination.""" def walk( self, top: str | Path, top_down: bool = True, on_error: Callable[[OSError], Any] | None = None, follow_symlinks: bool = False, ) -> AsyncIterator[tuple[str, list[str], list[str]]]: """Asynchronously walk a directory tree yielding (dirpath, dirnames, filenames).""" AsyncFile An asynchronous file handle wrapping a native file object, offering non-blocking reads and writes. class AsyncFile: @property def name(self) -> str: """The file path.""" @property def mode(self) -> str: """The open mode.""" @property def closed(self) -> bool: """Returns True if the file has been closed.""" @property def encoding(self) -> str | None: """File encoding (None for binary files).""" async def read(self, size: int = -1) -> bytes | str: """Read up to size bytes/characters. Reads all if size=-1.""" async def readline(self) -> bytes | str: """Read a single line.""" async def readlines(self) -> list[bytes | str]: """Read all lines into a list.""" async def readinto(self, buffer: bytearray) -> int: """Read bytes into a pre-allocated buffer (binary mode only).""" async def write(self, data: bytes | str) -> int: """Write data to the file.""" async def writelines(self, lines: Iterable[bytes | str]) -> None: """Write an iterable of lines.""" async def seek(self, offset: int, whence: int = 0) -> int: """Change the stream position.""" async def tell(self) -> int: """Return the current stream position.""" async def truncate(self, size: int | None = None) -> int: """Truncate the file to at most size bytes.""" async def flush(self) -> None: """Flush write buffer to disk.""" async def close(self) -> None: """Flush write buffers and close the file handle.""" AsyncPath An asynchronous object-oriented path class wrapper providing pathlib-style methods. class AsyncPath: def __init__(self, *args: str | Path | AsyncPath, **kwargs: Any) -> None @property def name(self) -> str: ... @property def stem(self) -> str: ... @property def suffix(self) -> str: ... @property def parent(self) -> AsyncPath: ... async def exists(self) -> bool: ... async def is_file(self) -> bool: ... async def is_dir(self) -> bool: ... async def stat(self) -> os.stat_result: ... async def mkdir(self, parents: bool = True, exist_ok: bool = True) -> None: ... async def rmdir(self) -> None: ... async def unlink(self, missing_ok: bool = False) -> None: ... async def open(self, mode: str = "r", **kwargs: Any) -> AsyncFile: ... async def read_text(self, encoding: str = "utf-8") -> str: ... async def write_text(self, data: str, encoding: str = "utf-8") -> None: ... async def read_bytes(self) -> bytes: ... async def write_bytes(self, data: bytes) -> None: ... Filesystem Fault Hierarchy Exception faults raised by the async filesystem module under the io domain. Fault Class Fault Code Description ))} )
Go to Homepage