Storage Effect — Aquilia Documentation
Comprehensive guide and documentation for Storage Effect in the Aquilia framework. View API reference, examples, and implementation patterns.
EFFECTS / UNIFIED STORAGE Storage Effect The StorageEffect provides unified file and object storage operations. Managed by the StorageProvider , it abstracts filesystem locations and cloud storage backends (AWS S3, Google Cloud Storage, Azure Blobs) into a standard API. Storage Abstraction Layer Hardcoding file paths or importing third-party cloud SDKs directly into handlers ties your application logic to a specific cloud provider. The StorageEffect decouples this connection. Handlers operate on simple keys (e.g., "invoices/invoice_123.pdf") inside a scoped bucket namespace, leaving the backend storage configuration to the provider setup. StorageHandle API The acquired handle is an instance of StorageHandle . It manages binary data transfer safely across local filesystems and cloud buckets: await handle.read(key: str) -> bytes | None Reads file contents as raw bytes. Returns None if the file is missing or connection issues arise. await handle.write(key: str, data: bytes) -> None Writes raw bytes to the specified key. Overwrites existing contents if present. Automatically constructs directory trees if needed on a local filesystem. await handle.delete(key: str) -> bool Deletes the file matching the key. Returns True if deletion was successful, False otherwise. await handle.exists(key: str) -> bool Queries the backend to check if a file exists under the key. Returns a boolean status. Usage: Profile Avatar Secure Uploader The example below demonstrates receiving a profile avatar, generating a unique filename, writing it to S3 via the StorageHandle , and saving the record reference to the database: import uuid from aquilia.controller import Controller, POST, RequestCtx from aquilia.flow import requires from aquilia.effects import StorageEffect, DBTx class AvatarUploadController(Controller): # Require both storage bucket access and database transaction effects = [ StorageEffect("user-avatars"), DBTx["write"] ] @POST("/profile/avatar") async def upload_avatar(self, ctx: RequestCtx) -> dict: storage = ctx.get_effect("Storage") # StorageHandle instance db = ctx.get_effect("DBTx") # DBTxHandle instance # 1. Fetch file from multipart request body uploaded_file = await ctx.request.file("avatar") if not uploaded_file: return ctx.json( , status=400) # 2. Validate file extension ext = uploaded_file.filename.split(".")[-1].lower() if ext not in ("jpg", "jpeg", "png", "webp"): return ctx.json( , status=400) # 3. Generate a secure, unique filename key secure_key = f"profiles/ / . " # 4. Write bytes to the cloud storage bucket await storage.write(secure_key, uploaded_file.content) # 5. Update user avatar reference in database await db.execute( "UPDATE profiles SET avatar_path = ? WHERE user_id = ?", (secure_key, ctx.user.id) ) return HTTP Effect Custom Effects )
Go to Homepage