Integration — Aquilia Documentation
Comprehensive guide and documentation for Integration in the Aquilia framework. View API reference, examples, and implementation patterns.
Contracts / Controller Integration Controller Integration Contracts integrate deeply with Aquilia's Controllers, DI Container, Auth system, and Sessions. This page shows how Contracts work as part of the full request lifecycle. Auto-Binding to Request When a controller handler declares a Contract type hint, Aquilia automatically parses the request body and creates a Contract instance via bind_contract_to_request(): from aquilia import Controller, Post, Put, Get from myapp.contracts import ProductContract class ProductController(Controller): prefix = "/api/products" @Post("/", status_code=201) async def create(self, ctx, payload: ProductContract): """ 'payload' is automatically: 1. Parsed from request JSON/form body 2. Instantiated as ProductContract(data=body) 3. Context injected with request + DI container """ if not payload.is_sealed(): return ctx.json(payload.errors, status=422) product = payload.imprint() await product.save() return ctx.json(ProductContract(instance=product).data, status=201) @Put("/ ") async def update(self, ctx, id: int, payload: ProductContract): product = await Product.objects.get(id=id) if not payload.is_sealed(): return ctx.json(payload.errors, status=422) payload.imprint(instance=product, partial=True) await product.save() return ctx.json(ProductContract(instance=product).data) Response Rendering with Projections from aquilia import Controller, Get from myapp.contracts import ProductContract class ProductController(Controller): prefix = "/api/products" @Get("/") async def list_products(self, ctx): products = await Product.objects.all() # Minimal projection for list views bp = ProductContract( instance=products, many=True, projection="__minimal__" ) return ctx.json(bp.data) @Get("/ ") async def detail(self, ctx, id: int): product = await Product.objects.get(id=id) # Full detail projection bp = ProductContract(instance=product, projection="detail") return ctx.json(bp.data) @Get("/ /admin") async def admin_view(self, ctx, id: int): product = await Product.objects.get(id=id) # Admin projection with all fields bp = ProductContract(instance=product, projection="__all__") return ctx.json(bp.data) DI Container Integration Contracts can resolve dependencies from the DI container using the Inject facet: from aquilia.contracts import Contract, TextFacet, Inject, Hidden from aquilia.di import Singleton # Register a service in DI class PricingService: def calculate_tax(self, price: float, region: str) -> float: rates = return price * rates.get(region, 0.10) # In workspace.py: container.singleton(PricingService) class InvoiceContract(Contract): amount = FloatFacet() region = TextFacet(max_length=10) # Inject service from DI container pricing = Inject(token=PricingService) # Hidden — populated by context, never exposed in I/O created_by_id = Hidden() class Spec: model = Invoice fields = ["amount", "region"] def seal_tax_calculation(self, data): """Use injected service for validation.""" tax = self.pricing.calculate_tax(data["amount"], data["region"]) if tax > 1000: self.reject("amount", "Invoice amount exceeds tax threshold.") Auth Integration Combine Contracts with Aquilia's auth guards for secure endpoints: from aquilia import Controller, Post, Get from aquilia.auth import guard, AuthGuard from myapp.contracts import ArticleContract class ArticleController(Controller): prefix = "/api/articles" @Post("/", status_code=201) @guard(AuthGuard) # Require authentication async def create(self, ctx, payload: ArticleContract): """Create article — requires authenticated user.""" if not payload.is_sealed(): return ctx.json(payload.errors, status=422) # Inject the authenticated user as the author article = payload.imprint() article.author = ctx.user # From auth guard await article.save() return ctx.json(ArticleContract(instance=article).data, status=201) @Get("/") async def list_articles(self, ctx): """Public endpoint — no auth required.""" articles = await Article.objects.filter(published=True) return ctx.json( ArticleContract( instance=articles, many=True, projection="public" ).data ) # Contract with auth-aware validation: class AdminSettingsContract(Contract): maintenance_mode = BoolFacet(default=False) max_upload_size = IntFacet(min_value=1, max_value=100) class Spec: model = Settings fields = ["maintenance_mode", "max_upload_size"] async def async_seal_admin_only(self, data): """Only admins can change maintenance mode.""" if data.get("maintenance_mode") and not self.context.get("is_admin"): self.reject("maintenance_mode", "Only admins can enable maintenance mode.") Sessions Integration Contracts can access session data for validation and conditional logic: from aquilia import Controller, Post from aquilia.sessions import session, authenticated from myapp.contracts import CartContract class CartController(Controller): prefix = "/api/cart" @Post("/checkout") @authenticated # Require authenticated session async def checkout(self, ctx): """Checkout flow using session data + Contract.""" # Get cart items from session cart_items = ctx.session.get("cart", []) # Validate checkout data with Contract bp = CheckoutContract(data= ) if not bp.is_sealed(): return ctx.json(bp.errors, status=422) order = bp.imprint() order.user = ctx.session.principal await order.save() # Clear cart from session ctx.session.delete("cart") return ctx.json(OrderContract(instance=order).data, status=201) class CheckoutContract(Contract): items = ListFacet(child=IntFacet(), min_items=1) shipping_address = NestedContractFacet(AddressContract) class Spec: model = Order fields = ["items", "shipping_address"] def seal_cart_not_empty(self, data): if not data.get("items"): self.reject("items", "Your cart is empty.") Full CRUD Example from aquilia import Controller, Get, Post, Put, Patch, Delete from aquilia.auth import guard, AuthGuard from myapp.contracts import ProductContract class ProductController(Controller): prefix = "/api/products" @Get("/") async def list(self, ctx): products = await Product.objects.all() return ctx.json( ProductContract(instance=products, many=True, projection="__minimal__").data ) @Get("/ ") async def retrieve(self, ctx, id: int): product = await Product.objects.get(id=id) return ctx.json(ProductContract(instance=product, projection="detail").data) @Post("/", status_code=201) @guard(AuthGuard) async def create(self, ctx, payload: ProductContract): if not await payload.is_sealed_async(): return ctx.json(payload.errors, status=422) product = payload.imprint() await product.save() return ctx.json(ProductContract(instance=product).data, status=201) @Patch("/ ") @guard(AuthGuard) async def partial_update(self, ctx, id: int, payload: ProductContract): product = await Product.objects.get(id=id) # partial=True: only validate provided fields bp = ProductContract(data=await ctx.json(), partial=True) if not bp.is_sealed(): return ctx.json(bp.errors, status=422) bp.imprint(instance=product, partial=True) await product.save() return ctx.json(ProductContract(instance=product).data) @Delete("/ ", status_code=204) @guard(AuthGuard) async def destroy(self, ctx, id: int): product = await Product.objects.get(id=id) await product.delete() return ctx.empty() Utility Functions from aquilia.contracts.integration import ( bind_contract_to_request, render_contract_response, resolve_contract_from_annotation, is_contract_class, is_projected_contract, ) # Check if a class is a Contract is_contract_class(ProductContract) # True is_contract_class(str) # False # Check if it's a projected Contract reference is_projected_contract(ProductContract["summary"]) # True # Resolve Contract from a type annotation bp_class = resolve_contract_from_annotation(handler_param_annotation) # Returns (ContractClass, projection_name) or None # Bind Contract to an incoming request (used internally by controller engine) bp = await bind_contract_to_request(ProductContract, request, container) # Render model data through Contract for response data = render_contract_response(ProductContract, instance, projection="summary") )
Go to Homepage