Your API receives a file. A user hits ‘process’. Synchronously, the endpoint waits for the full operation to complete. For small workloads, this works fine. For large-scale AI pipelines, it becomes a bottleneck. The request times out. The user sees a spinner.
Process it asynchronously instead: the endpoint queues the job and responds immediately. A separate worker pool handles the heavy lifting in parallel. Results stream back in real-time via Server-Sent Events. The API stays responsive. This is the difference between an MVP and a product that scales.
This is the architecture every AI-integrated startup in the UAE needs: FastAPI as the API layer, a task queue (Celery or Dramatiq) for background work, and Server-Sent Events (SSE) to push results back to the client without polling. We’ll build this pattern step-by-step, with connection pooling, error handling, and a deployment story.
Why Async Task Queues Matter for AI Pipelines
Document processing, embedding generation, vector database enrichment, and model inference are expensive operations. Running them in the request/response cycle limits throughput. A single long-running request ties up a worker thread. Ten concurrent requests exhaust your pool. Your API becomes a bottleneck.
An async task queue decouples work from the request. The API queues the job and returns immediately. A separate worker pool processes jobs in parallel, independent of API load. If a job takes 30 seconds, the API doesn’t care. If you need more throughput, you scale workers independently from API replicas.
For UAE startups building RAG systems, document ingestion platforms, or AI enrichment services, this pattern is non-negotiable. It’s what separates a prototype from production.
The Architecture: Three Layers
Picture this:
- API Layer (FastAPI): Receives requests, queues jobs, returns job IDs, streams results via SSE.
- Message Broker (Redis or RabbitMQ): Stores job messages. Celery or Dramatiq pull from it.
- Worker Layer (Celery/Dramatiq): Processes jobs, calls your AI models, writes results to a database or cache.
The client connects to the API, gets a job ID, then connects to an SSE endpoint to stream results as they complete. No polling loops. No blocking waits.
Setting Up FastAPI with Celery
Start with dependencies:
pip install fastapi uvicorn celery redis pydantic python-dotenvCreate a Celery app:
from celery import Celery
from celery.result import AsyncResult
import os
celery_app = Celery(
"data_pipeline",
broker=os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0"),
backend=os.getenv("CELERY_BACKEND_URL", "redis://localhost:6379/1"),
)
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="UTC",
enable_utc=True,
)
@celery_app.task(bind=True, name="process_document")
def process_document(self, file_path: str, doc_id: str):
try:
# Simulate document processing
# In production: extract text, chunk, embed, store in vector DB
result = {"doc_id": doc_id, "status": "completed", "chunks": 42}
return result
except Exception as exc:
# Log error, update task state
self.update_state(state="FAILURE", meta={"error": str(exc)})
raiseNow the FastAPI endpoints:
from fastapi import FastAPI, UploadFile, File, BackgroundTasks
from fastapi.responses import StreamingResponse
import asyncio
import json
from typing import AsyncGenerator
app = FastAPI()
@app.post("/process")
async def submit_processing(file: UploadFile = File(...)):
"""Queue a document for processing, return job ID immediately."""
# Save file temporarily
file_path = f"/tmp/{file.filename}"
with open(file_path, "wb") as f:
f.write(await file.read())
# Queue the task
task = celery_app.send_task(
"process_document",
args=(file_path, file.filename),
task_id=f"{file.filename}_{int(time.time())}"
)
return {"job_id": task.id, "status": "queued"}
@app.get("/results/{job_id}")
async def stream_results(job_id: str) -> StreamingResponse:
"""Stream job results via Server-Sent Events."""
async def event_generator() -> AsyncGenerator[str, None]:
while True:
result = AsyncResult(job_id, app=celery_app)
if result.state == "PENDING":
data = {"status": "pending", "progress": 0}
elif result.state == "PROGRESS":
data = result.info
elif result.state == "SUCCESS":
data = {"status": "completed", "result": result.result}
yield f"data: {json.dumps(data)}
"
break
elif result.state == "FAILURE":
data = {"status": "failed", "error": str(result.info)}
yield f"data: {json.dumps(data)}
"
break
else:
data = {"status": result.state}
yield f"data: {json.dumps(data)}
"
await asyncio.sleep(1)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
)Run the worker in a separate process:
celery -A celery_app worker --loglevel=info --concurrency=4Connection Pooling: Reuse Connections, Don’t Create New Ones
If your tasks call a vector database or PostgreSQL, opening a fresh connection for every job degrades performance. Use connection pooling to reuse connections across tasks:
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
import asyncpg
# PostgreSQL pool
db_engine = create_engine(
"postgresql://user:pass@localhost/db",
poolclass=QueuePool,
pool_size=20,
max_overflow=10,
pool_recycle=3600,
)
# Vector DB (Pinecone, Weaviate, or self-hosted) connection pool
async def get_vector_db_pool():
return await asyncpg.create_pool(
"postgresql://user:pass@vector-db:5432/vectors",
min_size=10,
max_size=20,
)
@celery_app.task(name="embed_and_store")
def embed_and_store(text: str, doc_id: str):
"""Embed text and store in vector DB using pooled connection."""
# Get connection from pool (doesn't create new one)
with db_engine.connect() as conn:
# Embed using your model
embedding = model.encode(text)
# Store in vector DB
conn.execute(
"INSERT INTO embeddings (doc_id, embedding, text) VALUES (%s, %s, %s)",
(doc_id, embedding.tolist(), text),
)
conn.commit()
return {"doc_id": doc_id, "embedded": True}Connection pooling significantly reduces overhead by reusing connections instead of creating new ones for each task. This matters at scale.
Error Handling and Retry Logic
Network calls fail. Vector databases timeout. Your task needs resilience. Celery supports automatic retries with exponential backoff:
from celery.exceptions import MaxRetriesExceededError
import logging
logger = logging.getLogger(__name__)
@celery_app.task(
bind=True,
name="process_with_retry",
autoretry_for=(Exception,),
retry_kwargs={"max_retries": 3, "countdown": 5},
default_retry_delay=5,
)
def process_with_retry(self, data: dict):
try:
# Your processing logic
result = call_external_api(data)
return result
except TimeoutError as exc:
logger.warning(f"Timeout on attempt {self.request.retries}, retrying...")
raise self.retry(exc=exc, countdown=10) # Exponential backoff
except Exception as exc:
if self.request.retries < 3:
logger.error(f"Task failed, retry {self.request.retries + 1}")
raise self.retry(exc=exc, countdown=5 * (self.request.retries + 1))
else:
logger.critical(f"Task exhausted retries: {exc}")
raise MaxRetriesExceededError(str(exc))This pattern retries failed tasks with exponential backoff, logs everything, and fails gracefully after max retries. For production workloads, this is essential.
Real-Time Streaming with Server-Sent Events
SSE is simpler than WebSockets for one-way server-to-client updates. The client connects and listens:
// Frontend JavaScript
const eventSource = new EventSource(`/results/${jobId}`);
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log("Job status:", data);
if (data.status === "completed") {
console.log("Results:", data.result);
eventSource.close();
} else if (data.status === "failed") {
console.error("Job failed:", data.error);
eventSource.close();
}
};
eventSource.onerror = () => {
console.error("Connection lost");
eventSource.close();
};The server pushes updates. No polling. No unnecessary requests. The user sees progress in real-time.
Deployment to Azure Container Apps
Package your API and workers in separate containers. Create a docker-compose file locally, then deploy to Azure Container Apps:
version: "3.8"
services:
api:
build: .
ports:
- "8000:8000"
environment:
- CELERY_BROKER_URL=redis://redis:6379/0
- CELERY_BACKEND_URL=redis://redis:6379/1
depends_on:
- redis
command: uvicorn main:app --host 0.0.0.0 --port 8000
worker:
build: .
environment:
- CELERY_BROKER_URL=redis://redis:6379/0
- CELERY_BACKEND_URL=redis://redis:6379/1
depends_on:
- redis
command: celery -A celery_app worker --loglevel=info --concurrency=4
redis:
image: redis:7-alpine
ports:
- "6379:6379"For Azure, push images to Azure Container Registry, then create Container Apps for the API and worker with auto-scaling rules. Set replica count for the worker based on queue depth.
Monitoring and Observability
You can’t fix what you can’t see. Add basic monitoring with Prometheus:
from prometheus_client import Counter, Histogram, start_http_server
import time
# Metrics
task_counter = Counter("tasks_total", "Total tasks", ["name", "status"])
task_duration = Histogram("task_duration_seconds", "Task duration", ["name"])
@celery_app.task(name="monitored_task")
def monitored_task(data: dict):
start = time.time()
try:
result = do_work(data)
task_counter.labels(name="monitored_task", status="success").inc()
return result
except Exception as exc:
task_counter.labels(name="monitored_task", status="failure").inc()
raise
finally:
duration = time.time() - start
task_duration.labels(name="monitored_task").observe(duration)
# Expose metrics on /metrics for Prometheus
if __name__ == "__main__":
start_http_server(8001)Scrape these metrics with Prometheus, visualize in Grafana. Track task success rates, latencies, and queue depth. Set alerts when queue depth grows or error rates spike.
Scaling Considerations
As load grows, tune these levers:
- Worker concurrency: More workers equals more parallel tasks. Start with 4, scale based on CPU and I/O.
- Task timeout: Set explicit timeouts to prevent zombie tasks. Use Celery’s
time_limitparameter. - Queue priority: Route urgent jobs to a high-priority queue, batch jobs to a low-priority queue.
- Result retention: Don’t keep results forever. Celery can auto-delete old results. Set
result_expiresto 3600 seconds.
For a typical UAE startup processing 1000 documents per day, 4 workers with 2 concurrency each handles the load. As you scale to 10,000 documents, add workers and increase concurrency. Monitor queue depth and adjust.
Patterns to Avoid
Blocking I/O in tasks: If your task calls an API without async support, use requests with timeouts, not bare blocking calls. Better yet, use httpx with async/await.
Forgetting task idempotency: If a task retries, it might run twice. Design tasks to be idempotent: storing the same embedding twice should be a no-op, not a duplicate.
No dead-letter queue: Tasks that fail repeatedly should be captured for manual inspection. Add a dead-letter queue to track failures.
Tight coupling to Redis: Redis is great for development but consider RabbitMQ for production. It’s more robust and supports priority queues natively.
Putting It Together
Your startup now has a production-grade data pipeline. A user uploads a document. The API queues it in 10ms. A worker picks it up, extracts text, generates embeddings, stores them in your vector database. The client streams progress via SSE. Meanwhile, your API is free to handle the next request.
This is the foundation for scalable AI products. RAG systems, document enrichment, batch inference, real-time analytics, all built on the same async, queue-driven pattern. As your UAE startup grows from MVP to scale-up, this architecture scales with you.
Start small: FastAPI plus Celery plus Redis locally. Deploy to Azure Container Apps. Monitor with Prometheus. As traffic grows, add workers, tune concurrency, migrate to RabbitMQ if needed. The pattern remains consistent.
What’s the difference between Celery and Dramatiq?
Both are task queue libraries for Python. Celery is older, more feature-rich, and heavier. Dramatiq is newer, lighter, faster for simple use cases, and easier to reason about. For most UAE startups, Dramatiq is a solid choice. Celery wins if you need complex routing, priority queues, or task chaining.
Do I need Redis or can I use RabbitMQ?
Either works. Redis is simpler to set up and faster for small-to-medium scale. RabbitMQ is more robust, supports message persistence out of the box, and scales better when you have millions of tasks. Start with Redis. Migrate to RabbitMQ when queue reliability becomes critical.
How do I handle long-running tasks that take hours?
Set explicit task timeouts and break long work into smaller subtasks. A 4-hour embedding job becomes 100 subtasks of 2 minutes each. Chain them with Celery’s chain or group primitives. This way, if one subtask fails, you retry only that one, not the whole 4 hours.
Can I use FastAPI’s native async/await instead of Celery?
FastAPI’s async is for I/O-bound work: API calls, database queries. It doesn’t parallelize CPU-bound work like embedding generation or model inference. For those, you need a separate worker pool. Celery gives you that. Use FastAPI async for API endpoints, Celery for background jobs.
How do I prevent duplicate processing if a task retries?
Make tasks idempotent. Use a unique constraint on your database. For embeddings, use (doc_id, chunk_id) as the primary key. If the same task runs twice, the second insert fails silently or updates the existing row. In Celery, use task_id to make retries deterministic.
