Introduction

If an API call should enqueue work and return immediately, FastAPI can do that without introducing Celery or Redis on day one. The cleanest in-process design is an asyncio.Queue with exactly one worker task consuming jobs in FIFO order. That gives you sequential execution, predictable ordering, and a fast HTTP response path, as long as you understand that the queue lives only inside one application process.

Why BackgroundTasks alone is not a queue

FastAPI's BackgroundTasks runs work after the response is sent, which is useful for lightweight follow-up actions. It does not, by itself, give you a single global worker, job ordering, or shared backpressure. If two requests arrive together, two background tasks can run independently.

When you need "one job at a time," you want a real queue plus one consumer. In an async FastAPI app, asyncio.Queue is a natural fit.

Build a single-worker queue with lifespan startup

The pattern below starts one worker when the app boots, accepts jobs through POST /jobs, and lets clients check progress through GET /jobs/job_id.

1from contextlib import asynccontextmanager, suppress
2import asyncio
3import uuid
4
5from fastapi import FastAPI, HTTPException
6from pydantic import BaseModel
7
8
9class JobRequest(BaseModel):
10    message: str
11
12
13job_queue: asyncio.Queue[tuple[str, str]] = asyncio.Queue()
14job_status: dict[str, str] = {}
15
16
17async def process_job(job_id: str, message: str) -> None:
18    await asyncio.sleep(2)
19    print(f"processed {job_id}: {message}")
20
21
22async def worker() -> None:
23    while True:
24        job_id, message = await job_queue.get()
25        job_status[job_id] = "running"
26
27        try:
28            await process_job(job_id, message)
29            job_status[job_id] = "done"
30        except Exception:
31            job_status[job_id] = "failed"
32        finally:
33            job_queue.task_done()
34
35
36@asynccontextmanager
37async def lifespan(app: FastAPI):
38    worker_task = asyncio.create_task(worker())
39    try:
40        yield
41    finally:
42        worker_task.cancel()
43        with suppress(asyncio.CancelledError):
44            await worker_task
45
46
47app = FastAPI(lifespan=lifespan)
48
49
50@app.post("/jobs", status_code=202)
51async def enqueue_job(payload: JobRequest):
52    job_id = str(uuid.uuid4())
53    job_status[job_id] = "queued"
54    await job_queue.put((job_id, payload.message))
55    return {"job_id": job_id, "status": "queued"}
56
57
58@app.get("/jobs/{job_id}")
59async def get_job(job_id: str):
60    status = job_status.get(job_id)
61    if status is None:
62        raise HTTPException(status_code=404, detail="Job not found")
63    return {"job_id": job_id, "status": status}

This is single-threaded in the sense that one async worker processes one queued job at a time. Other requests can still be served while the queue worker awaits network or disk work, but the jobs themselves are serialized by the single consumer.

Keep the job function async and non-blocking

The queue stays responsive only if the worker does not block the event loop for long stretches. If a job is CPU-heavy or calls blocking libraries, move that part into a thread pool or process pool. For simple I/O-heavy tasks, an async function is enough.

1import asyncio
2
3
4def generate_report_sync(message: str) -> str:
5    import time
6    time.sleep(3)
7    return message.upper()
8
9
10async def process_job(job_id: str, message: str) -> None:
11    result = await asyncio.to_thread(generate_report_sync, message)
12    print(f"{job_id}: {result}")

Using asyncio.to_thread() keeps the FastAPI event loop from freezing while still preserving the one-job-at-a-time queue order.

Understand the process boundary

This design works only inside one FastAPI process. If you run Uvicorn or Gunicorn with multiple workers, each worker process gets its own in-memory queue and its own status dictionary. That breaks the idea of one global ordered queue.

So if you need strict single-worker semantics, run one application worker. If you later need durability, retries, or multiple machines, move the queue to an external broker such as Redis and use a proper task system.

That is also why this approach is best for lightweight internal workloads, not mission-critical job processing where jobs must survive restarts.

Common Pitfalls

The most common mistake is using BackgroundTasks and assuming it guarantees serialized execution. It does not. It simply schedules work after the response.

Another issue is running multiple ASGI worker processes. That creates multiple independent queues, which defeats the "single queue" requirement.

Blocking code inside the worker is another common problem. A long synchronous operation can stall the whole event loop if you do not move it to asyncio.to_thread() or an external worker.

Finally, remember that an in-memory queue loses pending jobs on process restart. If you need durability, this design is too small for the requirement.

Summary

• Use asyncio.Queue plus one worker task when you need sequential job processing after an API call.
• Start the worker with FastAPI lifespan so it is created once per process.
• Return 202 Accepted from the enqueue endpoint and track job status separately.
• Keep blocking job logic off the event loop, for example with asyncio.to_thread().
• Run only one app worker if you need one true in-process queue.