LLM integration in Django: views, streaming, Celery and cost tracking

LLM integration in Django usually starts with a single API call in a view, and then questions appear: how not to block a worker for a minute, how not to pay for repeated questions, how to track spend. Below is a working structure for a Django project with the OpenAI SDK: settings, a wrapper module with caching, regular and streaming views, a Celery background task and cost tracking in the admin. All models in the examples are called through the Zukko gateway at its OpenAI-compatible URL.

Installation and settings

pip install django openai celery redis

Add the client parameters and a Redis cache to settings.py (the django.core.cache.backends.redis.RedisCache backend is built into Django since 4.0):

import os

ZUKKO_API_KEY = os.environ["ZUKKO_API_KEY"]
ZUKKO_BASE_URL = "https://api.zukko.pro/v1"
ZUKKO_MODEL = os.environ.get("ZUKKO_MODEL", "claude-sonnet-5")

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
    }
}

CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"

Pass the key through an environment variable rather than storing it in settings.py. You can create it on the API keys↗ page; for a website it's convenient to have a separate key with a daily limit.

Wrapper module: one client, cache, token accounting

File myapp/llm.py:

import hashlib
import json
from decimal import Decimal

from django.conf import settings
from django.core.cache import cache
from openai import OpenAI

client = OpenAI(
    base_url=settings.ZUKKO_BASE_URL,
    api_key=settings.ZUKKO_API_KEY,
    timeout=60,
    max_retries=2,
)

What matters here:

• The client is created once when the module is imported and reuses connections. timeout=60 keeps a request from hanging forever, and max_retries=2 retries on network failures.
• The cache key is built from the model and all messages via json.dumps(..., sort_keys=True). A common mistake is concatenating strings like system or "" + user: due to operator precedence, when a system prompt is set, the user text never gets into the key, and every question receives the same cached answer.
• Prices come from the Zukko price list (dollars per 1M tokens) and are stored as strings so Decimal computes without rounding errors. Prices and the model set change: the current list is at GET https://api.zukko.pro/v1/models or on the "Models" page in your dashboard.

Besides the Django cache, Zukko has a server cache for fully identical requests: a repeat with the same body is served for free (see key limits, idempotency and cache↗). The Django cache additionally saves the network round trip.

A view with error handling

File myapp/views.py:

import logging

from django.http import JsonResponse, StreamingHttpResponse
from django.views.decorators.http import require_POST
from openai import APIConnectionError, APIStatusError

from .llm import ask

logger = logging.getLogger(__name__)

ERRORS = {
    401: "Configuration error: invalid API key",
    402: "Service temporarily unavailable: balance exhausted",
    403: "Model not allowed for this key",

The user gets a clear message and the details go to the log. Don't put raw str(e) into the response.

On the frontend, insert the answer with textContent, not innerHTML: both the user's message and the model's answer are untrusted text, otherwise you get XSS. The CSRF token is included automatically if you send the form's FormData with {% csrf_token %}.

Streaming response with StreamingHttpResponse

To make text appear as it is generated, pass a generator to StreamingHttpResponse. Add to myapp/llm.py:

def stream_answer(user_message, system=None, model=None):
    stream = client.chat.completions.create(
        model=model or settings.ZUKKO_MODEL,
        messages=build_messages(user_message, system),
        max_tokens=1024,
        stream=True,
    )
    for chunk in stream:
        if chunk.choices:
            text = chunk.choices[0].delta.content
            if text:
                yield text

And a view in myapp/views.py:

from .llm import stream_answer

@require_POST
def chat_stream(request):
    message = clean_message(request)
    if message is None:
        return JsonResponse({"error": "Empty or too long message"}, status=400)
    return StreamingHttpResponse(
        stream_answer(message), content_type="text/plain; charset=utf-8"
    )

The if chunk.choices: check is mandatory: the last chunk of the stream carries only usage and an empty choices list, and without the check chunk.choices[0] fails with IndexError at the end of the answer.

Keep in mind the limitations from the Django documentation: under WSGI a streaming response occupies a worker for the whole generation, so for long jobs it's better to move the work out of the request-response cycle. Under ASGI, StreamingHttpResponse must receive an async iterator — use AsyncOpenAI and async for for that. In the streaming variant, API errors surface after the headers are sent, inside the generator — log them there.

Background tasks with Celery

Long documents and batch processing shouldn't hold an HTTP request. Models in myapp/models.py:

from django.conf import settings
from django.db import models

class ProcessedText(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    input_text = models.TextField()
    result = models.TextField(blank=True)
    status = models.CharField(max_length=20, default="pending")
    error_message = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

class APIUsage(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True,
                             on_delete=models.SET_NULL)
    model_name = models.CharField(max_length=50)
    input_tokens = models.PositiveIntegerField()
    output_tokens = models.PositiveIntegerField()
    cost_usd = models.DecimalField(max_digits=12, decimal_places=8)
    created_at = models.DateTimeField(auto_now_add=True)

The task in myapp/tasks.py:

from celery import shared_task
from openai import APIConnectionError, APIStatusError, RateLimitError

from .llm import ask
from .models import ProcessedText

@shared_task(bind=True, max_retries=3)
def process_text(self, text_id):
    obj = ProcessedText.objects.get(pk=text_id)
    try:
        obj.result = ask(obj.input_text, use_cache=False, user=obj.user)
        obj.status = "completed"
    except (RateLimitError, APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=30)
    except APIStatusError as exc:
        obj.status = "failed"
        obj.error_message = f"{exc.status_code}: {exc.message}"
    obj.save(update_fields=["result", "status", "error_message"])

Only temporary errors (429 and network failures) are retried. Retrying 401, 402 or 403 is pointless — the task is marked as failed right away. In the view, create the object and start the task: process_text.delay(obj.pk). Connecting Celery to Django needs the standard celery.py from the Celery documentation, and the worker is started with celery -A myproject worker -l info.

Spend in the admin

from django.contrib import admin
from .models import APIUsage

@admin.register(APIUsage)
class APIUsageAdmin(admin.ModelAdmin):
    list_display = ("created_at", "user", "model_name",
                    "input_tokens", "output_tokens", "cost_usd")
    list_filter = ("model_name", "created_at")

Example: a claude-sonnet-5 request with 800 input and 300 output tokens costs (800 × 0.3 + 300 × 0.3) / 1,000,000 = $0.00033. You can cross-check these numbers with the "Usage" section in the dashboard. Basic Python client setup is covered in the Python guide↗.

FAQ

Why cache if requests are cheap?

Caching saves both money and time: a repeated answer arrives without calling the model. On a site with typical questions (an FAQ bot, form hints), the share of repeats can be noticeable — check your own logs.

Sync or async client?

Under WSGI use the sync OpenAI client as in the examples. If the project runs under ASGI and you write async views, use AsyncOpenAI with the same parameters.

Which model should I pick?

The example defaults to claude-sonnet-5; the cheapest is glm-5.3-flash ($0.04 per 1M tokens). The model is set by the ZUKKO_MODEL variable, so you can compare them on your own requests without changing code.

What if the Celery worker isn't running?

The task waits in the broker queue and runs when a worker starts. The pending status in the model helps show the user that processing is still in progress.