Controller Guide — Aquilia Documentation
Comprehensive guide and documentation for Controller Guide in the Aquilia framework. View API reference, examples, and implementation patterns.
Filesystem / Controller Guide Controller Integration Learn how to inject the FileSystem service, parse and validate file uploads, stream chunked files, check folder directories safely, and handle filesystem faults. Injecting the FileSystem Service The FileSystem service handles thread pool offloading and is automatically registered in the DI container. Simply declare FileSystem as a parameter type in your Controller constructor. Aquilia resolves it automatically using type annotations, without requiring explicit Inject() defaults: # modules/files/controllers.py from aquilia import Controller, GET, POST, RequestCtx, Response from aquilia.filesystem import FileSystem, FileNotFoundFault, PathTraversalFault class MediaController(Controller): prefix = "/media" # Auto-wired via DI using type annotations def __init__(self, fs: FileSystem): self.fs = fs self.storage_root = "./var/media" @GET("/read/ ") async def view_file(self, ctx: RequestCtx, filename: str) -> Response: # Resolve path safely inside our storage root boundary target_path = f" / " try: # FileSystem automatically validates target_path sandboxing content = await self.fs.read_file(target_path, sandbox=self.storage_root) return Response.text(content) except FileNotFoundFault: return Response.json( , status=404) except PathTraversalFault: return Response.json( , status=400) Handling File Uploads Process uploaded files safely by running name sanitization and writing bytes asynchronously: from aquilia.filesystem import sanitize_filename @POST("/upload") async def upload(self, ctx: RequestCtx) -> Response: file = await ctx.request.file("file") if not file: return Response.json( , status=400) # Sanitize name to avoid malicious shell/path segments safe_name = sanitize_filename(file.filename) dest_path = f" / " # Read uploaded stream content = await file.read() # Write atomically (dest_path parent is created if missing) await self.fs.write_file(dest_path, content, atomic=True, mkdir=True) return Response.json( , status=201) Directory Listing Browser Build a safe file list directory endpoint using scan_dir and validate_relative_path: from aquilia.filesystem import validate_relative_path, file_stat @GET("/list/ ") async def list_folder(self, ctx: RequestCtx, folder: str = "") -> Response: # Ensure relative path has no '..' segments if folder: try: folder = validate_relative_path(folder) except Exception: return Response.json( , status=400) full_path = f" / " if folder else self.storage_root try: # scan_dir returns a list of DirEntry dataclasses entries = await self.fs.scan_dir(full_path, sandbox=self.storage_root) results = [] for entry in entries: # DirEntry holds cached boolean flags entry_type = "directory" if entry.is_dir_cached else "file" results.append( ) return Response.json( ) except FileNotFoundFault: return Response.json( , status=404) Handling Filesystem Faults Catch concrete filesystem faults from aquilia.filesystem to return clean status codes: from aquilia.filesystem import ( FileSystemFault, FileNotFoundFault, PermissionDeniedFault, FileExistsFault, DiskFullFault ) @GET("/info") async def info(self, ctx: RequestCtx) -> Response: try: stat = await self.fs.file_stat(f" /config.json") return Response.json( ) except FileNotFoundFault: return Response.json( , status=404) except PermissionDeniedFault: return Response.json( , status=403) except DiskFullFault: return Response.json( , status=507) except FileSystemFault as e: return Response.json( "}, status=500) )
Go to Homepage