Controller Guide — Aquilia Documentation
Comprehensive guide and documentation for Controller Guide in the Aquilia framework. View API reference, examples, and implementation patterns.
SQLite / Controller Guide Controller Integration Learn how to execute queries, manage transaction scopes, handle constraint exceptions, and paginate records inside your HTTP Controllers using the SqliteService. Injecting the SqliteService The SqliteService manages the connection pool lifecycle (opening on startup, closing on shutdown) and is registered in the DI container automatically. Simply declare SqliteService as a parameter type in your Controller constructor. Aquilia resolves it automatically using type annotations, without requiring explicit Inject() defaults: # modules/users/controllers.py from aquilia import Controller, GET, POST, RequestCtx, Response from aquilia.sqlite import SqliteService, SqliteIntegrityError class UsersController(Controller): prefix = "/users" # Auto-wired via DI using type annotations def __init__(self, db: SqliteService): self.db = db @GET("/") async def list_users(self, ctx: RequestCtx) -> Response: # Access the ConnectionPool via self.db.pool rows = await self.db.pool.fetch_all("SELECT id, name, email FROM users") # Rows behave like dictionaries: users = [ for row in rows ] return Response.json(users) @POST("/") async def create_user(self, ctx: RequestCtx) -> Response: data = await ctx.json() try: async with self.db.pool.acquire(readonly=False) as conn: async with conn.transaction(): # execute returns AsyncCursor, allowing us to read lastrowid cursor = await conn.execute( "INSERT INTO users (name, email) VALUES (?, ?)", [data["name"], data["email"]] ) user_id = cursor.lastrowid return Response.json( , status=201) except SqliteIntegrityError: return Response.json( , status=400) Complete CRUD Operations A complete controller implementing standard CRUD logic, checking cursor results to return proper HTTP responses: from aquilia import Controller, GET, POST, PUT, DELETE, RequestCtx, Response from aquilia.sqlite import SqliteService, SqliteIntegrityError class ItemsController(Controller): prefix = "/items" def __init__(self, db: SqliteService): self.db = db @GET("/ ") async def get_item(self, ctx: RequestCtx, item_id: int) -> Response: row = await self.db.pool.fetch_one( "SELECT id, title, description FROM items WHERE id = ?", [item_id] ) if not row: return Response.json( , status=404) return Response.json(dict(row)) @PUT("/ ") async def update_item(self, ctx: RequestCtx, item_id: int) -> Response: data = await ctx.json() async with self.db.pool.acquire(readonly=False) as conn: async with conn.transaction(): cursor = await conn.execute( "UPDATE items SET title = ?, description = ? WHERE id = ?", [data["title"], data["description"], item_id] ) if cursor.rowcount == 0: return Response.json( , status=404) return Response.json( ) @DELETE("/ ") async def delete_item(self, ctx: RequestCtx, item_id: int) -> Response: async with self.db.pool.acquire(readonly=False) as conn: async with conn.transaction(): cursor = await conn.execute("DELETE FROM items WHERE id = ?", [item_id]) if cursor.rowcount == 0: return Response.json( , status=404) return Response.json( ) Query Pagination Paginate large SQL results by compiling count queries and using OFFSET parameters. @GET("/search") async def search_items(self, ctx: RequestCtx) -> Response: page = int(ctx.request.query.get("page", 1)) per_page = int(ctx.request.query.get("per_page", 10)) offset = (page - 1) * per_page # Run total count query count_val = await self.db.pool.fetch_val("SELECT COUNT(*) FROM items") total = count_val or 0 # Fetch paginated rows rows = await self.db.pool.fetch_all( "SELECT id, title FROM items LIMIT ? OFFSET ?", [per_page, offset] ) return Response.json( ) Handling SQLite Exceptions Catch custom SQLite exceptions to return specific, actionable API error responses. from aquilia.sqlite import ( SqliteIntegrityError, SqliteQueryError, PoolExhaustedError ) @POST("/safe-action") async def safe_action(self, ctx: RequestCtx) -> Response: try: await self.db.pool.execute("INSERT INTO audit_logs DEFAULT VALUES") return Response.json( ) except SqliteIntegrityError as e: return Response.json( "}, status=400) except SqliteQueryError as e: return Response.json( , status=500) except PoolExhaustedError: return Response.json( , status=503) )
Go to Homepage