Controller Guide — Aquilia Documentation
Comprehensive guide and documentation for Controller Guide in the Aquilia framework. View API reference, examples, and implementation patterns.
Unified Storage / Controller Guide Controller Integration Learn how to interact with the Storage Registry inside your HTTP Controllers. This guide covers file uploads, streaming downloads, signed URLs, and structured fault handling. Accessing the Registry The StorageRegistry singleton is automatically bound to the DI container when the workspace starts. You can inject it into your controllers directly. Simply declare StorageRegistry as a parameter type in your Controller constructor. Aquilia resolves it automatically using type annotations, without requiring explicit Inject() defaults: # modules/api/controllers.py from aquilia import Controller, GET, POST, RequestCtx, Response from aquilia.storage import StorageRegistry class FilesController(Controller): prefix = "/files" # Auto-wired via DI using type annotations def __init__(self, registry: StorageRegistry): self.registry = registry @POST("/upload") async def upload(self, ctx: RequestCtx) -> Response: file = await ctx.request.file("file") if not file: return Response.json( , status=400) content = await file.read() # Save using default backend saved_name = await self.registry.default.save(file.filename, content) return Response.json( ) Streaming Downloads To deliver large files efficiently without exhausting server RAM, open files as streams and read them in chunks using the chunks() generator. @GET("/download/ ") async def download(self, ctx: RequestCtx, filename: str) -> Response: backend = self.registry.default if not await backend.exists(filename): return Response.json( , status=404) metadata = await backend.stat(filename) async def file_sender(): # open returns a StorageFile supporting async chunk iteration async with await backend.open(filename) as f: async for chunk in f.chunks(chunk_size=1024 * 64): # 64KB chunks yield chunk return Response.stream( file_sender(), headers= "' } ) Signed / Presigned URLs For cloud storage backends (S3, GCS, Azure), generate temporary signed URLs to allow clients to download files directly from the cloud. @GET("/signed-url/ ") async def get_url(self, ctx: RequestCtx, filename: str) -> Response: backend = self.registry["s3"] # Fetch specific S3 backend alias if not await backend.exists(filename): return Response.json( , status=404) # Generate URL valid for 30 minutes (1800 seconds) url = await backend.url(filename, expire=1800) return Response.json( ) Handling Storage Faults Catch specific storage faults from aquilia.storage to return clean, appropriate HTTP status codes. from aquilia.storage import ( FileNotFoundError, PermissionError, StorageError ) @GET("/info/ ") async def file_info(self, ctx: RequestCtx, filename: str) -> Response: try: metadata = await self.registry.default.stat(filename) return Response.json(metadata.to_dict()) except FileNotFoundError: # Raised by stat() if key does not exist return Response.json( , status=404) except PermissionError: # Lacking local read rights or cloud IAM permissions return Response.json( , status=403) except StorageError as e: # Base class for other storage faults return Response.json( "}, status=500) )
Go to Homepage