Skip to Content
ContributeBackendBackground Tasks

Background Tasks

The backend offloads long-running work (test execution, generation, Architect chat) to Celery. The API enqueues a task and returns immediately with a task ID; the worker runs it. This page covers the backend/API side — enqueuing from a route and polling status. For worker internals (Celery config, BaseTask, tenant-context decorator, running and monitoring workers), see Worker Background Tasks.

Enqueuing from a route

Launch tasks with task_launcher, which pulls organization_id and user_id off current_user so they don’t have to be passed explicitly, then returns the Celery AsyncResult:

route_handler.py
from rhesis.backend.tasks import task_launcher, execute_test_configuration

@router.post("/{test_configuration_id}/execute")
def execute_test_configuration_endpoint(
    test_configuration_id: UUID,
    current_user: schemas.User = Depends(require_current_user_or_token),
):
    result = task_launcher(
        execute_test_configuration,
        str(test_configuration_id),
        current_user=current_user,
    )
    return {"task_id": result.id}

The task inherits tenant context from the launch call, so downstream database work runs scoped to the right organization without threading IDs through by hand.

Polling status

Clients poll GET /jobs/\{task_id\}, which reads the result backend:

job.py
@router.get("/{task_id}")
async def get_task_status(task_id: uuid.UUID, current_user=Depends(require_current_user_or_token)):
    result = AsyncResult(str(task_id), app=celery_app)
    return {
        "task_id": str(task_id),
        "status": result.status,
        "result": result.result if result.ready() else None,
        "error": str(result.error) if result.failed() else None,
    }

Callers submit, receive a task_id, then poll until status is SUCCESS or FAILURE. Some endpoints (e.g. test-set generation, endpoint exploration) also emit progress over WebSocket — see Architect Chat System.

Multi-tenancyEmail Notifications