Developer Guide — Aquilia Documentation
Comprehensive guide and documentation for Developer Guide in the Aquilia framework. View API reference, examples, and implementation patterns.
Developer Integration Guide Connecting DI, Controllers, Contracts, Models, Storage, Cache, and Mail This guide outlines how to build a unified application flow in Aquilia. You will learn how to connect Dependency Injection , HTTP Controllers , validation Contracts , ORM Models , Storage , Cache , and Mail in a single cohesive codebase. 1. Dependency Injection & Service Scopes Mark classes as services using the @service decorator. Constructor parameters are automatically autowired by the container using type hints: from aquilia.di import service from aquilia.cache import CacheService from aquilia.storage import StorageBackend # 1. Registered as a singleton across the entire app scope @service(scope="app") class ProductCatalogService: def __init__(self, cache: CacheService, storage: StorageBackend): self.cache = cache self.storage = storage # 2. Registered once per HTTP request scope and disposed on request end @service(scope="request") class UserContext: def __init__(self): self.user = None 2. Contracts: Schema & Input Validation A Contract defines request data validation schemas. Use the @ward decorator to implement custom cross-field constraints: from aquilia.contracts import Contract, Field, ward class UserRegistrationContract(Contract): email: str = Field(max_length=255) password: str = Field(min_length=8) password_confirm: str = Field() username: str = Field(min_length=3, max_length=50) @ward def validate_password_match(self) -> None: if self.password != self.password_confirm: raise ValueError("Passwords do not match") 3. Models: Database & Transactions Database models inherit from the base Model class. Wrap multiple queries inside atomic() async context managers to execute database transactions: from aquilia.models import Model, CharField, AutoField, atomic class User(Model): id = AutoField(primary_key=True) username = CharField(unique=True, max_length=50) email = CharField(unique=True, max_length=255) password_hash = CharField(max_length=255) avatar_url = CharField(null=True, max_length=512) async def create_user_transaction(user_data: dict, password_hash: str) -> User: # Executes queries inside a database transaction block async with atomic(): user = await User.objects.create( username=user_data["username"], email=user_data["email"], password_hash=password_hash ) return user 4. Putting It All Together: The Unified Controller Here is how to orchestrate a complete flow inside an HTTP Controller . The endpoint binds schema contracts, saves files to StorageBackend , writes transaction records, invalidates related CacheService entries, and dispatches custom TemplateMessage emails: from aquilia import Controller, POST from aquilia.controller import RequestCtx from aquilia.http import Response from aquilia.cache import CacheService from aquilia.storage import StorageBackend from aquilia.mail import TemplateMessage from .contracts import UserRegistrationContract from .models import User, create_user_transaction class RegistrationController(Controller): prefix = "/users" # Dependency Injection automatically wires service parameters def __init__(self, cache: CacheService, storage: StorageBackend): self.cache = cache self.storage = storage @POST("/register") async def register(self, ctx: RequestCtx): # 1. Bind and validate request body against Contract schema contract contract = await ctx.bind(UserRegistrationContract) # 2. Process file uploads via Storage avatar_file = ctx.request.files.get("avatar") avatar_url = None if avatar_file: filename = f"avatars/ .png" await self.storage.save(filename, avatar_file.read()) avatar_url = self.storage.url(filename) # 3. Write data inside a database transaction user = await create_user_transaction(contract.to_dict(), "hashed_pw") if avatar_url: user.avatar_url = avatar_url await user.save() # 4. Invalidate related cache tags await self.cache.delete("catalog:count") # 5. Dispatch template confirmation email msg = TemplateMessage( template="welcome.aqt", context= , subject="Welcome, >!", to=[user.email] ) await msg.asend() return Response.json(user.to_dict(), status=201) Quick Start Introduction )
Go to Homepage