Function enqueue_job

python/celery/main.py:88–118

(body: CreateJobRequest)

@app.post("/api/jobs", status_code=status.HTTP_201_CREATED)
def enqueue_job(body: CreateJobRequest):
    if body.task not in TASK_REGISTRY:
        raise HTTPException(status_code=400, detail=f"Unknown task: {body.task}")

    celery_task, params_model = TASK_REGISTRY[body.task]

    try:
        validated = params_model(**body.params)
    except ValidationError as exc:
        msg = exc.errors()[0].get("msg", "Invalid parameters")
        raise HTTPException(status_code=400, detail=msg) from exc

    params = validated.model_dump()
    job_id = uuid4().hex[:8]
    create_job(job_id, task=body.task, params=params)

    try:
        result = celery_task.apply_async(
            args=[job_id], kwargs=params, task_id=job_id, queue=QUEUE_NAME,
        )
    except Exception as exc:
        update_job(job_id, status="failed", result={"error": str(exc)})
        raise HTTPException(status_code=500, detail="Failed to enqueue task.") from exc

    update_job(job_id, message_id=str(result.id), queue_name=QUEUE_NAME)
    return {
        "id": job_id,
        "messageId": str(result.id),
        "queueName": QUEUE_NAME,
        "taskName": body.task,
    }