FastAPI Background Tasks vs. Celery for AI Feature Processing: When to Queue and When to Fire-and-Forget

from fastapi import BackgroundTasks, FastAPI
from sqlalchemy.orm import Session

app = FastAPI()

def log_ai_feature_usage(tenant_id: str, feature_name: str, db: Session):
 """Non-blocking usage logging—fires after response returns."""
 try:
 usage = FeatureUsage(
 tenant_id=tenant_id,
 feature_name=feature_name,
 timestamp=datetime.utcnow(),
 )
 db.add(usage)
 db.commit()
 except Exception as e:
 # This fails silently in BackgroundTasks. Genuinely doesn't matter here.
 logger.warning(f"Failed to log usage: {e}")

@app.post("/analyze")
async def analyze_document(
 tenant_id: str,
 document: str,
 background_tasks: BackgroundTasks,
 db: Session = Depends(get_db),
):
 # Return immediately while logging happens in background
 background_tasks.add_task(log_ai_feature_usage, tenant_id, "document_analysis", db)

 return {"status": "queued", "document_id": "..."}

import asyncio
from datetime import datetime

async def process_ai_inference(tenant_id: str, input_text: str, db: Session):
 """Process Claude API call asynchronously."""
 try:
 # Call Claude
 response = await anthropic_client.messages.create(
 model="claude-3-5-sonnet-20241022",
 max_tokens=1024,
 messages=[{"role": "user", "content": input_text}],
 )

 # Store result
 inference = AIInference(
 tenant_id=tenant_id,
 input=input_text,
 output=response.content[0].text,
 status="completed",
 completed_at=datetime.utcnow(),
 )
 db.add(inference)
 db.commit()
 except Exception as e:
 logger.error(f"Inference failed for tenant {tenant_id}: {e}")
 # Try to mark as failed, but if DB is down, this silently fails
 inference = AIInference(
 tenant_id=tenant_id,
 status="failed",
 error=str(e),
 )
 db.add(inference)
 db.commit()

@app.post("/infer")
async def infer(tenant_id: str, input_text: str, db: Session = Depends(get_db)):
 # Fire task without waiting
 asyncio.create_task(process_ai_inference(tenant_id, input_text, db))
 return {"status": "processing"}

# tasks.py
from celery import Celery, Task
from kombu import Queue

app = Celery("citiezenapp")
app.conf.update(
 broker_url="redis://localhost:6379/0",
 result_backend="redis://localhost:6379/0",
 task_serializer="json",
 accept_content=["json"],
 result_expires=3600,
)

# Per-tenant queue routing
app.conf.task_queues = (
 Queue("default", routing_key="default"),
 Queue("ai_inference", routing_key="ai.inference"),
)

class CallbackTask(Task):
 """Track task state in DB for multi-tenant visibility."""
 def on_failure(self, exc, task_id, args, kwargs, einfo):
 tenant_id = kwargs.get("tenant_id")
 logger.error(f"Task {task_id} failed for tenant {tenant_id}: {exc}")
 # Update tenant's failure count, trigger alert

@app.task(base=CallbackTask, bind=True, max_retries=3)
def process_ai_inference(
 self,
 tenant_id: str,
 inference_id: str,
 input_text: str,
):
 """
 Process Claude inference with automatic retries.
 Lives in Redis until completion.
 """
 try:
 db = next(get_db())
 inference = db.query(AIInference).filter_by(id=inference_id).first()

 if not inference:
 return {"error": "Inference not found"}

 # Check tenant rate limits (per-tenant concurrency)
 tenant_config = db.query(TenantAIConfig).filter_by(
 tenant_id=tenant_id
 ).first()

 if tenant_config.concurrent_inferences >= tenant_config.max_concurrent:
 # Retry after 10 seconds
 raise self.retry(countdown=10)

 inference.status = "processing"
 db.commit()

 response = anthropic_client.messages.create(
 model="claude-3-5-sonnet-20241022",
 max_tokens=1024,
 messages=[{"role": "user", "content": input_text}],
 )

 inference.output = response.content[0].text
 inference.status = "completed"
 inference.completed_at = datetime.utcnow()
 db.commit()

 return {"inference_id": inference_id, "status": "completed"}

 except Exception as exc:
 # Exponential backoff: 60s, 120s, 240s
 raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

# fastapi_app.py
@app.post("/infer")
async def infer(
 tenant_id: str,
 input_text: str,
 db: Session = Depends(get_db),
):
 inference = AIInference(
 tenant_id=tenant_id,
 input=input_text,
 status="queued",
 )
 db.add(inference)
 db.commit()

 # Queue task, returns immediately
 task = process_ai_inference.apply_async(
 kwargs={
 "tenant_id": tenant_id,
 "inference_id": inference.id,
 "input_text": input_text,
 },
 queue="ai_inference",
 priority=5, # Per-tenant prioritization
 )

 inference.celery_task_id = task.id
 db.commit()

 return {"inference_id": inference.id, "task_id": task.id}