Building a Django API that handles AI inference, processes long-running tasks, and isolates tenant data is fundamentally different from serving traditional CRUD endpoints. You need to think about request lifecycle, task orchestration, data boundaries, and response streaming. This guide covers the patterns we use in production at scale.
Async Views: Non-Blocking Request Handling
Django 4.1 introduced native async view support. For AI APIs, this matters because you want to avoid blocking the event loop while waiting for database queries, external API calls, or inference services. A blocked event loop means other requests queue up and timeouts cascade.
Here’s a typical pattern for an AI-powered endpoint that calls an inference service:
import asyncio
from django.http import JsonResponse
from asgiref.sync import sync_to_async
import httpx
@sync_to_async
def get_user_context(user_id):
"""Fetch user data from database."""
from myapp.models import User
return User.objects.get(id=user_id)
async def predict_endpoint(request):
"""Non-blocking AI prediction endpoint."""
try:
user = await get_user_context(request.user.id)
# Call inference service without blocking
async with httpx.AsyncClient() as client:
response = await client.post(
'https://inference-api.example.com/predict',
json={'input': request.POST.get('text')},
timeout=30.0
)
result = response.json()
return JsonResponse({'prediction': result['output']})
except asyncio.TimeoutError:
return JsonResponse(
{'error': 'Inference timeout'},
status=504
)
except Exception as e:
return JsonResponse(
{'error': str(e)},
status=500
)
The key here is using sync_to_async for database operations and httpx.AsyncClient for external calls. This keeps the event loop free for other requests. Use this pattern for endpoints where the response is ready within a few seconds.
For long-running inferences that take 10, 30, or 60+ seconds, don’t block the request. Return a task ID and let Celery handle the work. The search results confirm this: if the user must wait more than a few seconds, move the work out of the request cycle entirely.
Celery Task Queues for Background Processing
AI inference often takes seconds or minutes. Keeping the client waiting isn’t practical. Celery lets you queue tasks and return immediately with a job ID. The client polls or subscribes to updates while the work happens in the background.
Set up Celery with Django:
# celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_TIME_LIMIT = 600 # 10 minutes hard limit
CELERY_TASK_SOFT_TIME_LIMIT = 540 # 9 minutes soft limit
Now define a task for AI inference:
# tasks.py
from celery import shared_task
from myapp.models import InferenceJob
import requests
import json
@shared_task(bind=True, max_retries=3)
def run_inference(self, job_id, tenant_id):
"""Run AI inference and store results."""
try:
job = InferenceJob.objects.get(id=job_id, tenant_id=tenant_id)
job.status = 'processing'
job.save()
# Call inference service
response = requests.post(
'https://inference-api.example.com/predict',
json={'input': job.input_text},
timeout=300
)
response.raise_for_status()
result = response.json()
job.output = json.dumps(result)
job.status = 'completed'
job.save()
return {'job_id': job_id, 'status': 'completed'}
except requests.RequestException as exc:
# Retry with exponential backoff
raise self.retry(exc=exc, countdown=2 ** self.request.retries)
except InferenceJob.DoesNotExist:
# Task called with invalid job_id
return {'error': 'Job not found'}
In your view, queue the task and return the job ID immediately:
# views.py
from django.http import JsonResponse
from myapp.models import InferenceJob
from myapp.tasks import run_inference
def create_inference(request):
"""Create an inference job and queue it."""
job = InferenceJob.objects.create(
tenant_id=request.user.tenant_id,
input_text=request.POST.get('text'),
status='queued'
)
# Queue the task
run_inference.delay(job.id, request.user.tenant_id)
return JsonResponse({
'job_id': str(job.id),
'status': 'queued'
})
def get_inference_result(request, job_id):
"""Fetch inference result."""
try:
job = InferenceJob.objects.get(
id=job_id,
tenant_id=request.user.tenant_id
)
return JsonResponse({
'job_id': str(job.id),
'status': job.status,
'result': job.output
})
except InferenceJob.DoesNotExist:
return JsonResponse(
{'error': 'Job not found'},
status=404
)
PostgreSQL Row-Level Security for Multi-Tenant Isolation
In a multi-tenant SaaS, data isolation is critical. Row-Level Security (RLS) ensures that queries automatically filter by tenant, preventing accidental cross-tenant leaks. This is a database-level safety net that catches bugs your application logic might miss.
Set up RLS on your inference jobs table:
-- Create tenant table
CREATE TABLE tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(255) NOT NULL,
created_at TIMESTAMP DEFAULT NOW()
);
-- Create inference jobs table with tenant reference
CREATE TABLE inference_jobs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
input_text TEXT NOT NULL,
output JSONB,
status VARCHAR(50) DEFAULT 'queued',
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- Enable RLS
ALTER TABLE inference_jobs ENABLE ROW LEVEL SECURITY;
-- Create policy: users can only see their tenant's jobs
CREATE POLICY tenant_isolation ON inference_jobs
FOR ALL
USING (tenant_id = current_setting('app.current_tenant_id')::uuid);
-- Create policy for admin access
CREATE POLICY admin_access ON inference_jobs
FOR ALL
USING (current_setting('app.is_admin') = 'true');
In your Django middleware, set the tenant context for each request:
# middleware.py
from django.db import connection
from django.utils.deprecation import MiddlewareMixin
class TenantMiddleware(MiddlewareMixin):
def process_request(self, request):
if request.user.is_authenticated:
tenant_id = request.user.tenant_id
is_admin = request.user.is_admin
# Set PostgreSQL session variables
with connection.cursor() as cursor:
cursor.execute(
"SET app.current_tenant_id = %s",
[str(tenant_id)]
)
cursor.execute(
"SET app.is_admin = %s",
['true' if is_admin else 'false']
)
Now any query on the inference_jobs table automatically filters by tenant. If a user tries to access another tenant’s data, PostgreSQL blocks it at the database level. This is defense in depth: your application checks tenant_id in the view, and the database enforces it again.
WebSocket Streaming for Real-Time AI Responses
For some AI workloads, you want to stream tokens as they’re generated rather than waiting for the full response. Django Channels handles WebSocket connections and real-time messaging.
First, install Django Channels and configure it:
pip install channels channels-redis
Set up your ASGI configuration:
# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.security.websocket import AllowedHostsOriginValidator
from myapp.consumers import InferenceConsumer
from django.urls import path
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
ws_urlpatterns = [
path('ws/inference/<str:job_id>/', InferenceConsumer.as_asgi()),
]
application = ProtocolTypeRouter({
'http': get_asgi_application(),
'websocket': AllowedHostsOriginValidator(
AuthMiddlewareStack(
URLRouter(ws_urlpatterns)
)
),
})
Create a WebSocket consumer that streams inference results:
# consumers.py
import json
import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer
from asgiref.sync import sync_to_async
from myapp.models import InferenceJob
import httpx
class InferenceConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.job_id = self.scope['url_route']['kwargs']['job_id']
self.user = self.scope['user']
self.job_group_name = f'inference_{self.job_id}'
# Verify user owns this job
job = await self.get_job()
if not job or job.tenant_id != self.user.tenant_id:
await self.close()
return
await self.channel_layer.group_add(
self.job_group_name,
self.channel_name
)
await self.accept()
async def disconnect(self, close_code):
await self.channel_layer.group_discard(
self.job_group_name,
self.channel_name
)
async def receive(self, text_data):
data = json.loads(text_data)
if data.get('action') == 'start':
await self.start_inference()
async def start_inference(self):
"""Stream inference results to client."""
try:
async with httpx.AsyncClient() as client:
async with client.stream(
'POST',
'https://inference-api.example.com/predict/stream',
json={'input': 'user input'},
timeout=300.0
) as response:
async for line in response.aiter_lines():
if line:
await self.send(text_data=json.dumps({
'type': 'token',
'data': line
}))
await self.send(text_data=json.dumps({
'type': 'complete',
'data': None
}))
except Exception as e:
await self.send(text_data=json.dumps({
'type': 'error',
'data': str(e)
}))
@sync_to_async
def get_job(self):
try:
return InferenceJob.objects.get(id=self.job_id)
except InferenceJob.DoesNotExist:
return None
Performance Tuning and Deployment
Here’s what we’ve learned running this in production:
- Connection pooling: Use PgBouncer for PostgreSQL. Django’s default connection handling doesn’t scale well with many concurrent async tasks. A connection pool prevents exhaustion under load.
- Monitor task queues: Use Flower to visualize Celery task execution and spot bottlenecks. Watch queue depth and worker utilization.
- Set task timeouts: AI inference can hang. Hard limits prevent workers from getting stuck. Set both soft and hard timeouts in your Celery config.
- Cache results: Avoid re-running the same prediction twice. Store results keyed by input hash and tenant ID.
- Separate queues: Use different Celery queues for different task types. High-priority inference on one queue, batch processing on another. This prevents slow batch jobs from blocking real-time requests.
- Deploy with ASGI: Use Gunicorn with an ASGI worker class or Daphne for WebSocket support. WSGI won’t handle async views or WebSockets.
Example Gunicorn command for async workers:
gunicorn config.wsgi:application \
--worker-class uvicorn.workers.UvicornWorker \
--workers 4 \
--worker-connections 1000 \
--bind 0.0.0.0:8000
For WebSockets, use Daphne:
daphne -b 0.0.0.0 -p 8000 config.asgi:application
Conclusion
Building production Django APIs for AI SaaS requires orchestrating async views, background tasks, and secure multi-tenant data isolation. The patterns here form the foundation: async views for fast endpoints, Celery for long-running work, PostgreSQL RLS for tenant boundaries, and WebSocket streaming for real-time responses.
Start with async views for endpoints that complete within seconds. Add Celery when tasks exceed a few seconds. Implement RLS early to prevent data leaks. WebSocket streaming comes later when you need real-time feedback. This progression keeps your codebase manageable while you scale.
When should I use async views vs. Celery tasks?
Use async views for operations that complete within seconds (database queries, external API calls). Use Celery for anything longer, especially AI inference. Async views free up the event loop; Celery frees up the request entirely, letting clients get immediate responses.
Does PostgreSQL RLS replace application-level tenant checks?
No, use both. RLS is a safety net at the database layer. Application-level checks (verifying tenant_id in your Django view) catch bugs earlier and provide better error messages. RLS prevents disasters when application logic fails.
How do I scale Celery for high inference volume?
Use a task broker like Redis or RabbitMQ, deploy multiple Celery workers, and monitor queue depth with Flower. For extreme scale, consider dedicated inference services outside Django, with Celery just orchestrating calls to them.
Can I use Django’s built-in async features instead of Celery?
Django async views are great for I/O-bound work, but they still tie up a worker process. For long-running tasks (AI inference), Celery is cleaner because it decouples task execution from the request lifecycle. You can return to the client immediately.
What if my inference service goes down?
Celery’s retry mechanism with exponential backoff handles transient failures. For persistent failures, set a max retry count and log the error. Store the job status in the database so the client can check later. Consider circuit breaker patterns for dependent services.
