Operations — Aquilia Documentation
Comprehensive guide and documentation for Operations in the Aquilia framework. View API reference, examples, and implementation patterns.
Filesystem / Guide File & Directory Operations A comprehensive guide to managing files, directory structures, streaming chunks, path globbing, and atomic operations. Directory Management Listing Directory Contents (list_dir) Acquires listing array of filenames (names only, not absolute paths). from aquilia.filesystem import list_dir # Returns list[str] of direct child names (files/folders) items = await list_dir("./modules") print(items) # ['core', 'users', 'auth'] Scanning Directory with Metadata (scan_dir) Retrieves list of DirEntry objects with cached status details. from aquilia.filesystem import scan_dir entries = await scan_dir("./data") for entry in entries: if entry.is_file_cached: print(f"File: at ") elif entry.is_dir_cached: print(f"Directory: ") Creating Directories (make_dir) Create a folder structure on the host disk safely. from aquilia.filesystem import make_dir # parents=False (raises if parent missing), exist_ok=False (raises if already exists) by default await make_dir("./uploads/images", parents=True, exist_ok=True) Removing Directories Delete empty directories or recursively prune folder trees. from aquilia.filesystem import remove_dir, remove_tree # 1. remove_dir - Deletes empty directory (raises if not empty) await remove_dir("./temp_folder") # 2. remove_tree - Deletes directory and all contents recursively # Silently ignore deletion exceptions if path does not exist await remove_tree("./cache_files", ignore_errors=True) Temporary Files & Directories Create self-cleaning files and folders using secure, context-managed wrappers. from aquilia.filesystem import async_tempfile, async_tempdir # 1. Temporary File Context async with async_tempfile(suffix=".csv", prefix="export-") as tmp: # tmp is an AsyncFile handle open in w+b mode await tmp.write(b"id,name\\n1,Alice") await tmp.flush() print(f"Temporary file created at: ") # File is closed and unlinked automatically here # 2. Temporary Directory Context async with async_tempdir() as tmpdir: # tmpdir is an AsyncPath object await (tmpdir / "manifest.json").write_text(' ') print(f"Temporary directory path: ") # Directory and all nested files deleted recursively on block exit File Locking Utilize cross-process advisory locks to coordinate access to files. from aquilia.filesystem import AsyncFileLock, read_file, write_file # Acquire exclusive write lock (blocks until acquired) async with AsyncFileLock("db.lock"): data = await read_file("db.json") # Mutate data await write_file("db.json", data) # Acquire lock with a timeout threshold (raises LockAcquisitionError on timeout) lock = AsyncFileLock("process.lock", timeout=5.0) try: async with lock: # Exclusive execution block pass except LockAcquisitionError: print("Could not acquire lock, proceeding to fallback") Chunk Streaming Stream files in binary chunks to avoid high RAM usage on large files: from aquilia.filesystem import stream_read, stream_copy # Stream read a file (yields bytes chunks) async for chunk in stream_read("archive.tar.gz", chunk_size=1024 * 64): process_chunk(chunk) # Stream copy directly from source to destination bytes_copied = await stream_copy("large.mp4", "backup.mp4", chunk_size=1024 * 1024) Path Globbing Search directory trees for files matching glob rules using the AsyncPath model: from aquilia.filesystem import AsyncPath root = AsyncPath("./src") # Find all python files recursively async for filepath in root.glob("**/*.py"): print(f"Found code: (Parent: )") )
Go to Homepage