Controller Guide — Aquilia Documentation
Comprehensive guide and documentation for Controller Guide in the Aquilia framework. View API reference, examples, and implementation patterns.
Background Tasks / Controller Guide Controller Integration Learn how to dispatch background tasks from HTTP Controllers, inject the TaskManager instance, track job states, and test your async logic. Dependency Injection Setup You do not need to manually register the TaskManager class inside your app manifests. When TasksIntegration is registered in workspace.py, the dependency injection container automatically binds the active manager as an app-scoped singleton. Simply declare TaskManager as a parameter type in your Controller constructor. Aquilia resolves it automatically using type annotations, without requiring explicit Inject() defaults: # modules/core/controllers.py from aquilia import Controller, POST, GET, RequestCtx, Response from aquilia.tasks import TaskManager from .tasks import send_welcome_email class SignupController(Controller): # Auto-wired via DI using type annotations def __init__(self, manager: TaskManager): self.manager = manager @POST("/signup") async def signup(self, ctx: RequestCtx) -> Response: data = await ctx.json() # Option A: Dispatch via injected manager (allows advanced options) job_id = await self.manager.enqueue( send_welcome_email, email=data["email"], priority=Priority.HIGH ) # Option B: Dispatch via descriptor delay helper directly # (TaskManager is auto-bound to send_welcome_email descriptor) job_id = await send_welcome_email.delay(email=data["email"]) return Response.json( ) Task Dependency Injection Aquilia's worker automatically resolves task parameters using the DI container based on type hints. Task functions do not need default Inject() markers. # modules/core/tasks.py from aquilia.tasks import task from modules.core.services import EmailService from aquilia.sqlite import SqliteService @task(queue="emails") async def send_welcome_email( email: str, email_service: EmailService, db: SqliteService, ): # DI resolves EmailService and SqliteService when the worker executes this task user = await db.fetch_one("SELECT name FROM users WHERE email = ?", [email]) name = user["name"] if user else "Valued User" await email_service.send_welcome(email, name) Tracking Job Status Provide status endpoints to allow clients to poll for background job outcomes or trace failures. from aquilia import Controller, GET, RequestCtx, Response from aquilia.tasks import TaskManager, JobState class JobStatusController(Controller): prefix = "/jobs" def __init__(self, manager: TaskManager): self.manager = manager @GET("/ ") async def status(self, ctx: RequestCtx, job_id: str) -> Response: job = await self.manager.get_job(job_id) if not job: return Response.json( , status=404) data = # Include outcomes if finished if job.state == JobState.COMPLETED: data["result"] = job.result.value data["duration_ms"] = job.result.duration_ms elif job.state in (JobState.FAILED, JobState.DEAD): data["error"] = job.result.error data["error_type"] = job.result.error_type data["retry_count"] = job.retry_count return Response.json(data) Testing Task Controllers To test controller routing without actually processing jobs asynchronously, mock the TaskManager . To test the task handler itself, run it synchronously as a standard coroutine. # tests/test_tasks.py import pytest from unittest.mock import AsyncMock from aquilia.tasks import TaskManager from modules.core.tasks import send_welcome_email @pytest.mark.asyncio async def test_controller_dispatches(client, mock_container): # Mock the TaskManager mock_manager = AsyncMock(spec=TaskManager) mock_manager.enqueue.return_value = "mock_job_123" mock_container.register_instance(TaskManager, mock_manager) response = client.post("/signup", json= ) assert response.status_code == 200 assert response.json()["job_id"] == "mock_job_123" @pytest.mark.asyncio async def test_task_logic_directly(): # Execute task synchronously by passing mocks directly mock_service = AsyncMock() mock_db = AsyncMock() mock_db.fetch_one.return_value = await send_welcome_email( email="test@example.com", email_service=mock_service, db=mock_db ) mock_service.send_welcome.assert_called_once_with("test@example.com", "Alice") Best Practices , , , ].map((item, i) => ( ))} )
Go to Homepage