Sync to Async: Migrating FastAPI Endpoints to arq/Redis

The problem: a PDF ingest that blocks for minutes

My ingest endpoint was synchronous. Upload a file, wait for OCR, wait for the LLM extraction pipeline, get a response. On a small note it took 3–5 seconds. On a dense PDF it could run for minutes. The original code even had a comment explaining the awkwardness:

@router.post("/ingest", response_model=list[IngestResult])
def ingest(file: UploadFile) -> list[IngestResult]:
    # Sync endpoint on purpose: the ingestion pipeline calls anyio.run()
    # internally (LLM extraction), which can't nest inside the request loop.

That comment is a smell. "Can't nest inside the request loop" means the work is too heavy for the request loop in the first place. Time to pull it out.

arq was already in pyproject.toml (arq>=0.26) and Redis was available locally. I just needed to wire it up.

Step 1: Probe the arq API before writing anything

Before touching production code I spent two minutes confirming the exact arq 0.28 surface I'd rely on:

from arq import create_pool, ArqRedis
from arq.connections import RedisSettings
from arq.jobs import Job, JobStatus, JobResult

print('JobStatus:', [s.value for s in JobStatus])
# JobStatus: ['deferred', 'queued', 'in_progress', 'complete', 'not_found']
print('from_dsn ok:', RedisSettings.from_dsn('redis://localhost:6379/0'))

Knowing the exact status strings matters because they flow through to the frontend polling loop and the OpenAPI schema.

Step 2: The worker module

I created src/pacos/platform/worker.py. The key pieces are the task function and WorkerSettings:

"""Arq background worker — runs long ingestion jobs off the request path.

Ingesting a PDF is OCR + LLM-heavy (seconds to minutes), which is too long to
block an HTTP request. The API enqueues a job here and returns a job id; the
client polls GET /api/knowledge/ingest/status/{job_id}.
"""

async def run_ingest(ctx: dict, filename: str, data: bytes) -> list[dict]:
    """Entry point called by arq. ctx is injected by the worker pool."""
    results = ingest_bytes(filename, data)   # the existing sync pipeline
    return [r.model_dump() for r in results]


class WorkerSettings:
    functions = [run_ingest]
    redis_settings = RedisSettings.from_dsn(settings.redis_url)
    max_jobs = 4
    job_timeout = 600  # 10 min ceiling for the worst PDFs

The task itself is thin: it just calls the existing ingest_bytes pipeline and returns serialisable dicts. All the complexity stays in the pipeline; the worker is just the carrier.

Step 3: Wiring the Redis pool into FastAPI's lifespan

The arq pool needs to live as long as the app and be shared across requests. FastAPI's lifespan context manager is the right place:

@asynccontextmanager
async def lifespan(_app: FastAPI):
    from pacos.shared.db import init_db
    init_db()

    # Arq pool — best-effort; if Redis is down we continue without it
    # (endpoints that need it will 503 explicitly).
    try:
        pool = await create_pool(RedisSettings.from_dsn(settings.redis_url))
        _app.state.arq_pool = pool
    except Exception:
        _app.state.arq_pool = None

    yield

    if _app.state.arq_pool:
        await _app.state.arq_pool.close()

Best-effort startup matters: during tests and offline dev I don't want a hard crash if Redis isn't reachable. Endpoints that actually enqueue jobs will 503; everything else keeps working.

Step 4: A clean dependency for the pool

Instead of reaching into request.app.state inside every route, I added one dependency to deps.py so routers stay clean:

from arq import ArqRedis
from fastapi import Depends, HTTPException, Request, status

def arq_pool(request: Request) -> ArqRedis:
    """The arq Redis pool set up at startup. 503 if Redis was unavailable."""
    pool = getattr(request.app.state, "arq_pool", None)
    if pool is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Job queue unavailable",
        )
    return pool

ArqPool = Annotated[ArqRedis, Depends(arq_pool)]

That Annotated alias is the same pattern I use for DbSession and CurrentUser. Routers just declare what they need; the DI container handles the wiring.

Step 5: The endpoint pair

The old single endpoint becomes two:

class IngestJob(BaseModel):
    job_id: str
    status: str  # always "queued" on return


@router.post("/ingest", response_model=IngestJob, status_code=202)
async def ingest(file: UploadFile, pool: ArqPool) -> IngestJob:
    """Accept a file, enqueue it, return a job handle immediately."""
    data = await file.read()
    job = await pool.enqueue_job("run_ingest", file.filename, data)
    return IngestJob(job_id=job.job_id, status="queued")


@router.get("/ingest/status/{job_id}")
async def ingest_status(job_id: str, pool: ArqPool) -> dict:
    """Poll for job progress. Frontend polls this until status == 'complete'."""
    job = Job(job_id, pool)
    status = await job.status()
    if status == JobStatus.not_found:
        raise HTTPException(status_code=404, detail="Job not found")
    if status == JobStatus.complete:
        info = await job.result_info()
        return {
            "status": "complete",
            "success": info.success,
            "result": info.result if info.success else None,
            "error": str(info.result) if not info.success else None,
        }
    return {"status": status.value}

The 202 Accepted status code is deliberate: the work hasn't finished, you got a receipt.

Step 6: The test suite problem I didn't anticipate

When I ran the existing API tests after wiring up the lifespan, they passed — but took 103 seconds instead of the usual 3. The lifespan was hitting Redis connection retries on every test run because the sandbox can't reach my local Redis.

Fix: patch out the pool in the test fixture:

# conftest.py / fixture
monkeypatch.setattr(app.state, "arq_pool", None)

With that, the suite dropped back to 3.45s and all 9 tests stayed green. The 503 behavior is correct: tests that exercise the ingest endpoint directly mock at a higher level.

Step 7: Frontend polling

On the client side I added two types and a polling helper to api.ts:

export interface IngestJob {
  job_id: string;
  status: string;
}

export type IngestPhase = "queued" | "in_progress" | "complete" | "error";

// In the api object:
ingest: (file: File) => upload<IngestJob>("/api/knowledge/ingest", form),
ingestStatus: (jobId: string) => get<IngestStatus>(
  `/api/knowledge/ingest/status/${jobId}`
),

The ingest page's doIngest handler became:

1. POST the file → get job_id back
2. Set phase to "queued"
3. setInterval polling ingestStatus(job_id) every 2s
4. On complete, clear the interval and refresh the concept list
5. On error, surface the message

The UI shows Queued… → Processing… → Done instead of a frozen spinner.

What the architecture looks like now

POST /ingest  ──► enqueue_job() ──► Redis queue
    ↓ 202 + job_id

GET /ingest/status/{id}  ◄── poll every 2s ◄── frontend
    ↓ {status: "in_progress"}
    ↓ {status: "complete", result: [...]}

arq worker process:
    dequeue ──► run_ingest() ──► LLM pipeline ──► write result to Redis

The API server never blocks. The worker is a separate process (arq pacos.platform.worker.WorkerSettings) that scales independently. The dependency injection pattern didn't change — routers just got a new ArqPool dependency alongside the existing DbSession.

The biggest lesson: don't wait until something is genuinely broken to pull heavy work off the request path. The blocking endpoint "worked fine" until it didn't. The arq integration turned out to be less than a day's work once I already had Redis available — most of the time went to the test fixture fix and the frontend polling loop, not the queue wiring itself.

---

Drafted by Claude Sonnet from my own Claude Code session transcript, then reviewed and edited before publishing.