Filesystem — Aquilia Documentation
Comprehensive guide and documentation for Filesystem in the Aquilia framework. View API reference, examples, and implementation patterns.
Filesystem Filesystem Module High-performance native async file I/O. A drop-in replacement for aiofiles that uses thread pool execution with atomic writes, streaming, and built-in security. Quick Example from aquilia.filesystem import FileSystem, async_open, read_file, write_file # Simple file operations content = await read_file("config.json") await write_file("output.txt", "Hello, World!") # Async file handle async with await async_open("data.csv", "r") as f: async for line in f: process(line) # Using FileSystem service (DI-injectable) fs = FileSystem() files = await fs.list_dir("./logs") for file in files: stats = await fs.stat(file) print(f" : bytes") Key Features ))} API Overview The module provides standalone functions and a FileSystem service class: Function Description ) })} Atomic Writes By default, write_file() performs atomic writes to prevent data corruption: # This is safe even if the process crashes mid-write await write_file("config.json", new_config) # How it works internally: # 1. Write to temp file: config.json.tmp.abc123 # 2. Sync to disk (fsync) # 3. Atomic rename: config.json.tmp.abc123 → config.json # Result: Either old or new content, never partial # Disable atomic writes if needed (not recommended) await write_file("log.txt", data, atomic=False) Why this matters: Without atomic writes, a crash during write can leave a file with partial content (truncated or mixed old/new data). Atomic writes guarantee the file is always in a consistent state. Security Features All path operations are automatically validated for security at the lowest layer: Path Traversal Protection Strict validation rejects any paths containing parent traversal patterns (e.g., ..) to keep file reads sandboxed. Null Byte Rejection Detects and raises a fault on C-style null bytes (\x00) to prevent extension spoofing and file truncation exploits. Path Length Limits Applies length constraints on inputs (defaulting to 4096 characters) to stop denial of service buffer overflow attempts. Filename Sanitization The sanitize_filename() utility strips dangerous control tags and symbols to sanitize client uploads. from aquilia.filesystem import read_file, validate_path, sanitize_filename from aquilia.filesystem import PathTraversalFault # These will raise PathTraversalFault: await read_file("../../../etc/passwd") # Traversal attack await read_file("file\\x00name.txt") # Null byte injection # Use validate_path for manual checking try: validate_path(user_provided_path) except PathTraversalFault: raise HTTPException(400, "Invalid path") # Sanitize user-provided filenames safe_name = sanitize_filename("my../file\\x00.txt") # "my_file_.txt" AsyncFile Handle For more control, use async_open() to get an async file handle: from aquilia.filesystem import async_open # Read mode async with await async_open("data.txt", "r") as f: content = await f.read() # Read all first_line = await f.readline() # Read one line lines = await f.readlines() # Read all lines # Write mode async with await async_open("output.txt", "w") as f: await f.write("Hello, ") await f.write("World!\\n") await f.writelines(["Line 1\\n", "Line 2\\n"]) # Binary mode async with await async_open("image.png", "rb") as f: data = await f.read() # Streaming iteration async with await async_open("large.csv", "r") as f: async for line in f: process_line(line) Why Not aiofiles? Feature aquilia.filesystem aiofiles ))} Installation The filesystem module is included in the core Aquilia package. No additional dependencies required. pip install aquilia # That's it! No additional packages needed. )
Go to Homepage