Celery redis setup and Jobs endpoint

Makefile

@@ -25,6 +25,7 @@ help:
 	@echo "make logs         - Stream all container logs"
 	@echo "make logs-app     - Stream app container logs"
 	@echo "make logs-ollama  - Stream Ollama container logs"
+	@echo "make logs-worker  - Stream Celery worker logs"
 	@echo "make shell        - Open shell in running app container"
 	@echo "make pull-model   - Pull Ollama model from .env.dev ($(OLLAMA_MODEL))"
 	@echo "make test         - Run test suite"
@@ -80,6 +81,9 @@ logs-app:
 logs-ollama:
 	@$(COMPOSE) logs -f ollama
 
+logs-worker:
+	@$(COMPOSE) logs -f celery-worker
+
 shell:
 	@$(COMPOSE) exec app /bin/sh
 

app/api/router.py

@@ -9,10 +9,11 @@
 
 from fastapi import APIRouter
 
-from app.api.routes import forms, templates
+from app.api.routes import forms, jobs, templates
 from app.api.v1.router import v1_router
 
 api_router = APIRouter()
 api_router.include_router(templates.router)
 api_router.include_router(forms.router)
 api_router.include_router(v1_router)
+api_router.include_router(jobs.router)

app/api/routes/jobs.py

@@ -0,0 +1,60 @@
+from fastapi import APIRouter, Depends
+from sqlmodel import Session
+
+from app.api.deps import get_db
+from app.api.schemas.forms import (
+    AsyncFormFill,
+    AsyncFormFillResponse,
+    AsyncJobSubmitResponse,
+    JobResponse,
+)
+from app.core.errors.base import AppError
+from app.db.repositories import create_job, get_job_by_uuid, get_template
+from app.models import Job
+from app.tasks.fill import fill_form_task
+
+router = APIRouter(tags=["jobs"])
+
+
+@router.get("/jobs/{job_id}", response_model=JobResponse)
+def get_job_status(job_id: str, db: Session = Depends(get_db)):
+    job = get_job_by_uuid(db, job_id)
+    if not job:
+        raise AppError("Job not found", status_code=404)
+    return JobResponse(
+        job_id=job.job_id,
+        job_type=job.job_type,
+        status=job.status,
+        progress_percent=job.progress_percent,
+        result_url=job.result_url,
+        error=job.error,
+        created_at=job.created_at.isoformat() if job.created_at else None,
+        updated_at=job.updated_at.isoformat() if job.updated_at else None,
+    )
+
+
+@router.post("/forms/jobs", response_model=AsyncFormFillResponse)
+def submit_async_form_fill(form: AsyncFormFill, db: Session = Depends(get_db)):
+    for tid in form.template_ids:
+        if not get_template(db, tid):
+            raise AppError(f"Template {tid} not found", status_code=404)
+
+    jobs: list[AsyncJobSubmitResponse] = []
+    for tid in form.template_ids:
+        result = fill_form_task.delay(tid, form.input_text, form.model)
+        job = Job(
+            celery_task_id=result.id,
+            job_type="form_generation",
+            template_id=tid,
+            input_text=form.input_text,
+            status="queued",
+            model=form.model,
+        )
+        job = create_job(db, job)
+        jobs.append(AsyncJobSubmitResponse(
+            job_id=job.job_id,
+            status=job.status,
+            poll_url=f"/api/v1/jobs/{job.job_id}",
+        ))
+
+    return AsyncFormFillResponse(jobs=jobs)

app/api/schemas/forms.py

@@ -1,9 +1,9 @@
 from pydantic import BaseModel, field_validator
 
+
 class FormFill(BaseModel):
     template_id: int
     input_text: str
-    # Optional Ollama model override for this fill; falls back to OLLAMA_MODEL.
     model: str | None = None
 
     @field_validator("input_text")
@@ -29,4 +29,49 @@ class TranscriptionResponse(BaseModel):
 
 class ModelsResponse(BaseModel):
     models: list[str]
-    default: str
+    default: str
+
+
+class AsyncFormFill(BaseModel):
+    template_ids: list[int]
+    input_text: str
+    model: str | None = None
+
+    @field_validator("input_text")
+    def validate_input_text(cls, value):
+        if not value or not value.strip():
+            raise ValueError("Input text cannot be empty")
+        return value
+
+    @field_validator("template_ids")
+    def validate_template_ids(cls, value):
+        if not value:
+            raise ValueError("template_ids cannot be empty")
+        return value
+
+
+class JobResponse(BaseModel):
+    job_id: str
+    job_type: str
+    status: str
+    progress_percent: int = 0
+    result_url: str | None = None
+    error: dict | None = None
+    created_at: str | None = None
+    updated_at: str | None = None
+
+    class Config:
+        from_attributes = True
+
+
+class AsyncJobSubmitResponse(BaseModel):
+    job_id: str
+    status: str
+    poll_url: str
+
+    class Config:
+        from_attributes = True
+
+
+class AsyncFormFillResponse(BaseModel):
+    jobs: list[AsyncJobSubmitResponse]

app/core/celery.py

@@ -0,0 +1,19 @@
+from celery import Celery
+
+from app.core.config import CELERY_BROKER_URL, CELERY_RESULT_BACKEND
+
+celery_app = Celery(
+    "fireform",
+    broker=CELERY_BROKER_URL,
+    backend=CELERY_RESULT_BACKEND,
+)
+
+celery_app.conf.update(
+    task_serializer="json",
+    result_serializer="json",
+    accept_content=["json"],
+    task_track_started=True,
+    result_expires=86400,
+)
+
+celery_app.conf.include = ["app.tasks.fill"]

app/core/config.py

@@ -37,6 +37,10 @@
 OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5:1.5b")
 WHISPER_HOST = os.getenv("WHISPER_HOST", "http://localhost:9000").rstrip("/")
 
+# --- Celery / Redis -------------------------------------------------------
+CELERY_BROKER_URL = os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0")
+CELERY_RESULT_BACKEND = os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/0")
+
 # --- CORS -----------------------------------------------------------------
 _DEFAULT_ORIGINS = "http://127.0.0.1:5173,http://localhost:5173"
 ALLOWED_ORIGINS = [

app/db/repositories.py

@@ -1,5 +1,5 @@
 from sqlmodel import Session, select
-from app.models import Template, FormSubmission
+from app.models import Template, FormSubmission, Job
 
 # Templates
 def create_template(session: Session, template: Template) -> Template:
@@ -22,3 +22,32 @@ def create_form(session: Session, form: FormSubmission) -> FormSubmission:
     session.commit()
     session.refresh(form)
     return form
+
+
+# Jobs
+def create_job(session: Session, job: Job) -> Job:
+    session.add(job)
+    session.commit()
+    session.refresh(job)
+    return job
+
+
+def get_job(session: Session, job_id: int) -> Job | None:
+    return session.get(Job, job_id)
+
+
+def get_job_by_uuid(session: Session, job_uuid: str) -> Job | None:
+    statement = select(Job).where(Job.job_id == job_uuid)
+    return session.exec(statement).first()
+
+
+def get_job_by_celery_id(session: Session, celery_task_id: str) -> Job | None:
+    statement = select(Job).where(Job.celery_task_id == celery_task_id)
+    return session.exec(statement).first()
+
+
+def update_job(session: Session, job: Job) -> Job:
+    session.add(job)
+    session.commit()
+    session.refresh(job)
+    return job

app/models/__init__.py

@@ -1,5 +1,5 @@
 """ORM models. Import from here: `from app.models import Template`."""
 
-from app.models.models import FormSubmission, Template
+from app.models.models import FormSubmission, Job, Template
 
-__all__ = ["Template", "FormSubmission"]
+__all__ = ["Template", "FormSubmission", "Job"]

app/models/models.py

@@ -1,5 +1,7 @@
-from sqlmodel import SQLModel, Field
+import uuid as uuid_mod
+
 from sqlalchemy import Column, JSON
+from sqlmodel import SQLModel, Field
 from datetime import datetime, timezone
 
 
@@ -16,4 +18,20 @@ class FormSubmission(SQLModel, table=True):
     template_id: int = Field(foreign_key="template.id")
     input_text: str
     output_pdf_path: str
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+
+
+class Job(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    job_id: str = Field(default_factory=lambda: str(uuid_mod.uuid4()), index=True, unique=True)
+    celery_task_id: str = Field(index=True)
+    job_type: str = Field(default="form_generation")
+    template_id: int | None = Field(default=None, foreign_key="template.id")
+    input_text: str | None = None
+    status: str = Field(default="queued")
+    progress_percent: int = Field(default=0)
+    result_url: str | None = None
+    error: dict | None = Field(default=None, sa_column=Column(JSON))
+    model: str | None = None
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

app/tasks/fill.py

@@ -0,0 +1,63 @@
+import logging
+from datetime import datetime, timezone
+
+from app.core.celery import celery_app
+from app.db.database import get_session
+from app.db.repositories import get_job_by_celery_id, get_template, update_job, create_form
+from app.models import FormSubmission
+from app.services.controller import Controller
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(bind=True, name="fill_form")
+def fill_form_task(self, template_id: int, input_text: str, model: str | None = None):
+    session = next(get_session())
+    try:
+        job = get_job_by_celery_id(session, self.request.id)
+        if not job:
+            raise RuntimeError(f"No job row for celery task {self.request.id}")
+
+        job.status = "processing"
+        job.progress_percent = 10
+        job.updated_at = datetime.now(timezone.utc)
+        update_job(session, job)
+
+        template = get_template(session, template_id)
+        if not template:
+            raise ValueError(f"Template {template_id} not found")
+
+        controller = Controller()
+        path = controller.fill_form(
+            user_input=input_text,
+            fields=template.fields,
+            pdf_form_path=template.pdf_path,
+            model=model,
+        )
+
+        submission = FormSubmission(
+            template_id=template_id,
+            input_text=input_text,
+            output_pdf_path=path,
+        )
+        create_form(session, submission)
+
+        job.status = "completed"
+        job.progress_percent = 100
+        job.result_url = f"/api/v1/forms/{submission.id}/download"
+        job.updated_at = datetime.now(timezone.utc)
+        update_job(session, job)
+
+        return {"job_id": job.job_id, "result_url": job.result_url}
+
+    except Exception as e:
+        logger.exception("fill_form_task failed")
+        job = get_job_by_celery_id(session, self.request.id)
+        if job:
+            job.status = "failed"
+            job.error = {"error_code": "TASK_FAILED", "message": str(e)}
+            job.updated_at = datetime.now(timezone.utc)
+            update_job(session, job)
+        raise
+    finally:
+        session.close()

docker/.env.example

@@ -20,6 +20,10 @@ OLLAMA_TIMEOUT=300
 WHISPER_HOST=http://whisper:9000
 WHISPER_MODEL=small.en
 
+# --- Celery / Redis -------------------------------------------------------
+CELERY_BROKER_URL=redis://redis:6379/0
+CELERY_RESULT_BACKEND=redis://redis:6379/0
+
 # --- Gunicorn (prod only) -------------------------------------------------
 GUNICORN_WORKERS=2