Django Recommender Search Backend Patterns

Experimental Django Recommender + Search Backend Best Practices

Implementation patterns for a Django backend serving mixed-results recommendations (Personalize / Databricks / microservice fan-out) and OpenSearch-backed search/feeds. 48 rules across 8 categories, ordered by execution lifecycle impact — earlier categories cascade through everything downstream.

This is the backend peer of the react-fetch-cache-patterns skill. React handles client-side waterfalls and caching; this skill handles server-side fan-out, downstream protection, OpenSearch query design, and ML-blend orchestration.

When to Apply

• Building or reviewing Django views that fan out to AWS Personalize, Databricks Model Serving, internal microservices, or any ML inference downstream
• Designing OpenSearch query endpoints (search results, infinite feeds, faceted search)
• Implementing a recommendations endpoint that blends multiple ranker outputs
• Investigating "Django backend slow when downstream is degraded" or "Personalize quota exhausted"
• Adding caching, retry, circuit breakers, or rate limiting to outbound calls
• Choosing between sync and async Django views, configuring uvicorn vs gunicorn
• Designing DRF response shapes for paginated feeds, partial results, or degraded paths

Rule Categories by Priority

Quick Reference

1. Fan-out Orchestration (CRITICAL)

• `orch-parallel-fanout-asyncio-gather` — Use asyncio.gather for independent downstream calls; never await sequentially
• `orch-return-exceptions-on-fanout` — return_exceptions=True so one failure doesn't poison the whole gather
• `orch-propagate-request-deadline` — Pass a deadline through every downstream call to bound whole-request latency
• `orch-reuse-async-clients` — One httpx.AsyncClient per downstream at module scope; never per-request
• `orch-bounded-fanout-concurrency` — Cap per-request fan-out with asyncio.Semaphore to protect the pool
• `orch-no-blocking-in-async-view` — Never block the event loop with sync ORM/IO in async views
• `orch-avoid-await-in-loop` — for item in items: await ... is serial; use asyncio.gather with comprehension
• `orch-batch-with-bulk-endpoint` — Bulk endpoint over N parallel calls; DataLoader pattern for batchers

2. External Service Protection (CRITICAL)

• `protect-per-downstream-timeout-budget` — Different timeouts per service matched to each downstream's p99
• `protect-circuit-breaker-per-downstream` — One breaker per downstream so failures stay isolated
• `protect-jittered-retry-backoff` — Full-jitter exponential backoff to prevent thundering-herd recovery
• `protect-no-retry-on-4xx` — Skip retry on 4xx and non-idempotent failures; distinguish connect vs read errors
• `protect-bulkhead-connection-pool` — One connection pool per downstream so one slow service doesn't starve others
• `protect-client-side-rate-limit` — Token bucket toward each downstream to stay under their quota
• `protect-honor-retry-after-header` — Parse Retry-After (seconds or HTTP-date) on 429/503

3. OpenSearch Query Patterns (CRITICAL)

• `search-use-search-after-not-from` — search_after cursor instead of from/size for any paginated endpoint
• `search-filter-source-fields` — Restrict _source to fields you render; use docvalue_fields for sortable
• `search-bool-filter-vs-must` — Non-scoring clauses in filter (cacheable), scoring clauses in must
• `search-function-score-for-blending` — Use function_score to blend personalization signals in-engine
• `search-stable-tiebreaker-sort` — Always end sort with _id (or unique numeric field) for stable cursors
• `search-alias-for-blue-green-reindex` — Query through aliases; never direct index names
• `search-enable-request-cache` — request_cache=true for hit-returning queries; canonicalize request body
• `search-shard-aware-routing` — Use routing keys to limit per-query shard fan-out

4. Result Blending & Personalization (HIGH)

• `blend-normalize-scores-across-sources` — Min-max or RRF normalize before blending Personalize/Databricks/OpenSearch
• `blend-mmr-for-diversity` — Maximal Marginal Relevance to avoid monocultures in top-K
• `blend-dedup-across-sources` — Canonical item ID dedup; bonus for cross-source corroboration
• `blend-cold-start-fallback` — Popular/editorial fallback for new users; tiered personalization
• `blend-anonymous-vs-personalized-paths` — Cheap segment-keyed cache for anon traffic; ML only for logged-in

5. Caching Strategy (HIGH)

• `cache-redis-with-stampede-protection` — SETNX lock + jittered TTL + probabilistic early refresh
• `cache-version-on-model-deploy` — Bake model version into cache keys; no flush needed on retrain
• `cache-segment-keyed-isolation` — Include auth/role/locale/segment in keys to prevent cross-context leakage
• `cache-two-tier-process-and-redis` — Process LRU in front of Redis for the hottest keys
• `cache-negative-results` — Cache absences and empty results with shorter TTL

6. Resilience & Partial Results (HIGH)

• `resilience-partial-response-envelope` — partial: true + sources_used + failed_sources in response
• `resilience-serve-stale-from-redis` — Two TTLs (fresh + stale); serve stale on origin failure
• `resilience-default-ranking-fallback` — Precomputed default ranking when all ML sources are down
• `resilience-per-source-observability` — Tag every downstream call with structured source/outcome metadata
• `resilience-degrade-search-gracefully` — Tier 1 → tier 2 → tier 3 fallback for OpenSearch outages

7. Async & Concurrency (MEDIUM-HIGH)

• `async-sync-to-async-orm` — Use Django 4.1+ async ORM (aget, afilter) or sync_to_async with thread_sensitive=True
• `async-worker-model-uvicorn-vs-gunicorn` — Run ASGI (uvicorn or gunicorn+UvicornWorker) for true async concurrency
• `async-fire-and-forget-with-create-task` — create_task for analytics/audit; add error handler; hold task references
• `async-context-vars-for-request-scope` — contextvars.ContextVar for per-request state; not threading.local
• `async-cancel-on-client-disconnect` — Check await request.is_disconnected(); propagate cancellation

8. API Response Design (MEDIUM)

• `api-cursor-pagination-in-drf` — Cursor pagination over page-number; opaque base64 cursors
• `api-serializer-perf-select-related` — select_related/prefetch_related/only to eliminate N+1
• `api-etag-and-cache-control-headers` — ETag + Cache-Control + Vary for CDN/client reuse
• `api-compression-and-payload-shaping` — gzip/brotli; sparse fieldsets; msgpack for internal APIs
• `api-throttle-per-user-and-endpoint` — DRF throttle classes per user/anon and per expensive endpoint

How to Use

1. Open references/_sections.md for category definitions and impact rationale 2. Read individual rule files for incorrect-vs-correct code examples (each ~150-300 lines with Python code) 3. For ready-to-use scaffolds, see scaffolding templates 4. The AGENTS.md navigation document (auto-generated) provides a TOC for browsing

Scaffolding Templates

Five ready-to-adapt Python templates under assets/templates/:

Reference Files

Related Skills

• react-fetch-cache-patterns — Client-side peer covering React data fetching/caching (Suspense, query libraries, prefetch)
• io-bound-data-processing — Python async patterns for batch and pipeline workloads
• inngest-nextjs-patterns — Workflow patterns for server-side step functions

Django Recommender + Search Backend

Version 0.1.0 Experimental May 2026

Note:

This document is mainly for agents and LLMs to follow when maintaining,

generating, or refactoring codebases. Humans may also find it useful,

but guidance here is optimized for automation and consistency by AI-assisted workflows.

---

Abstract

Implementation patterns for a Django backend API that serves mixed-results recommendations (fan-out to AWS Personalize, Databricks Model Serving endpoints, and internal microservices) and OpenSearch-backed search/feed endpoints. 46 rules across 8 categories ordered by execution lifecycle impact: Fan-out Orchestration (asyncio.gather, return_exceptions, deadline propagation, async client reuse, bounded concurrency, no-blocking-in-async, no-await-in-loop, DataLoader batching), External Service Protection (per-downstream timeouts, per-downstream circuit breakers, full-jitter retry, 4xx/non-idempotent retry policy, bulkhead pools, client-side rate limits, Retry-After parsing), OpenSearch Query Patterns (search_after vs from/size, _source filtering, bool.filter vs must, function_score for blending, stable tiebreaker sort, blue-green index aliases, request_cache, shard-aware routing), Result Blending & Personalization (score normalization, MMR diversity, canonical-ID dedup, cold-start fallback, anonymous-vs-personalized paths), Caching Strategy (Redis stampede protection via SETNX, model-version-keyed cache, segment-keyed isolation, two-tier process+Redis, negative result caching), Resilience & Partial Results (partial-response envelope, stale-on-error from Redis, default ranking fallback, per-source observability tags, tiered search degradation), Async & Concurrency (async ORM and sync_to_async, uvicorn/UvicornWorker deployment, create_task fire-and-forget, contextvars for request scope, cancel on client disconnect), and API Response Design (cursor pagination in DRF, select_related/prefetch_related/only, ETag + Cache-Control + Vary, gzip/brotli compression and payload shaping, throttling per user and endpoint). Bundled with 5 Python scaffolding templates: fanout_recommender_service, opensearch_search_view, result_blender, redis_cache_with_stampede, degraded_response. Backend peer to react-fetch-cache-patterns.

---

Table of Contents

1. Fan-out Orchestration — CRITICAL

• 1.1 Avoid await Inside Independent Loops — CRITICAL (reduces N sequential awaits to 1 round-trip time)
• 1.2 Batch Fan-out via Bulk Endpoints — CRITICAL (reduces N parallel calls to 1 round-trip)
• 1.3 Bound Per-Request Fan-out with a Semaphore — CRITICAL (prevents one user's request from saturating the pool)
• 1.4 Fan Out to Recommenders with asyncio.gather — CRITICAL (reduces N sequential downstream calls to 1 round-trip time)
• 1.5 Never Block the Event Loop in Async Views — CRITICAL (prevents 1 slow request from blocking all other requests)
• 1.6 Propagate a Request Deadline Across All Downstreams — CRITICAL (prevents unbounded request latency on slow downstream)
• 1.7 Reuse Async HTTP Clients Across Requests — CRITICAL (prevents 50-200ms per-request TLS handshake overhead)
• 1.8 Use return_exceptions=True for Partial-Results Fan-out — CRITICAL (prevents one downstream failure from failing the whole request)

2. External Service Protection — CRITICAL

• 2.1 Honor Retry-After Headers from Downstreams — HIGH (prevents 429 escalation and downstream bans)
• 2.2 Isolate Connection Pools per Downstream — HIGH (prevents one slow downstream from starving fast ones)
• 2.3 Run One Circuit Breaker per Downstream — CRITICAL (prevents one degraded downstream from cascading)
• 2.4 Set Per-Downstream Timeout Budgets — CRITICAL (prevents one slow service from blowing the whole budget)
• 2.5 Skip Retry on 4xx and Non-Idempotent Failures — HIGH (prevents wasted retries on permanent errors)
• 2.6 Throttle Outbound Calls with a Token Bucket — HIGH (prevents downstream rate-limit bans)
• 2.7 Use Full-Jitter Backoff for Server-to-Server Retries — CRITICAL (prevents thundering-herd recovery storms)

3. OpenSearch Query Patterns — CRITICAL

• 3.1 Blend Personalization Signals with function_score — HIGH (eliminates client-side re-rank of large candidate sets)
• 3.2 Enable request_cache for Repeat Queries — HIGH (10-100× faster for hot identical queries)
• 3.3 Include a Unique Tiebreaker in Every Sort — HIGH (prevents duplicate/missing items on paginated queries)
• 3.4 Paginate OpenSearch with search_after, Not from/size — CRITICAL (O(N) deep pagination → O(1))
• 3.5 Query Through Aliases, Never Direct Index Names — HIGH (enables zero-downtime reindex and rollback)
• 3.6 Restrict _source to Fields You Actually Render — CRITICAL (reduces search response size 10-100×)
• 3.7 Use bool.filter for Non-Scoring Clauses — CRITICAL (2-10× faster queries via filter cache)
• 3.8 Use Routing to Limit Shards Per Query — HIGH (10× faster queries when partition key known)

4. Result Blending & Personalization — HIGH

• 4.1 Apply MMR Diversity to Avoid Recommendation Monocultures — HIGH (prevents top-K from showing 10 near-duplicate items)
• 4.2 Dedup by Canonical ID Across All Sources — HIGH (prevents duplicate items in the final ranking)
• 4.3 Fall Back to Popular/Editorial on Cold-Start — HIGH (prevents empty recommendations for new users)
• 4.4 Normalize Scores Before Blending Across Sources — HIGH (prevents one source from dominating the ranking)
• 4.5 Separate Anonymous and Personalized Code Paths — MEDIUM-HIGH (prevents 60-80% of traffic hitting expensive personalization)

5. Caching Strategy — HIGH

• 5.1 Cache Negative Results to Prevent Origin Hammering — MEDIUM-HIGH (prevents N% origin traffic from invalid IDs and empty results)
• 5.2 Isolate Cache Keys by Segment and Auth State — HIGH (prevents cross-segment data leakage in cached responses)
• 5.3 Layer a Process LRU in Front of Redis — HIGH (reduces Redis RTT 30-90× for the hottest keys)
• 5.4 Protect Cache Misses from Stampede with a Lock — HIGH (prevents N concurrent regenerations on cold miss)
• 5.5 Version Cache Keys by Model Deploy — HIGH (prevents serving stale recommendations after model retrain)

6. Resilience & Partial Results — HIGH

• 6.1 Define a Default Ranking for Total ML Outage — HIGH (prevents empty recommendations when all sources fail)
• 6.2 Degrade Gracefully When OpenSearch Is Slow or Down — MEDIUM-HIGH (prevents search outages cascading to API outages)
• 6.3 Serve Stale Redis Data When Fresh Fetch Fails — HIGH (prevents transient downstream outages from breaking the API)
• 6.4 Surface Partiality in the Response Envelope — HIGH (prevents callers caching degraded responses as complete)
• 6.5 Tag Every Downstream Call with Structured Source Metadata — HIGH (prevents silent degradation in production)

7. Async & Concurrency — MEDIUM-HIGH

• 7.1 Cancel In-Flight Work When the Client Disconnects — MEDIUM (prevents wasted compute on abandoned requests)
• 7.2 Run Async Views Under Uvicorn or Gunicorn+UvicornWorker — MEDIUM-HIGH (enables true async concurrency per worker)
• 7.3 Use Async ORM Methods in Async Views — MEDIUM-HIGH (prevents event-loop blocking on ORM calls)
• 7.4 Use contextvars for Request-Scoped State Across Async Calls — MEDIUM (prevents cross-request state leakage in async code)
• 7.5 Use create_task for Fire-and-Forget Background Work — MEDIUM (prevents user requests blocking on analytics/audit writes)

8. API Response Design — MEDIUM

• 8.1 Apply Throttling per User and per Expensive Endpoint — MEDIUM (prevents one user exhausting expensive downstream quota)
• 8.2 Compress Responses and Shape Payloads — MEDIUM (reduces 60-80% of API egress bandwidth)
• 8.3 Return Cursor-Based Pagination from DRF — MEDIUM (prevents page-skip bugs as data shifts)
• 8.4 Set ETag and Cache-Control for CDN/Client Reuse — MEDIUM (enables 304 Not Modified responses and CDN caching)
• 8.5 Use select_related, prefetch_related, and only in DRF Serializers — MEDIUM (reduces N+1 queries from serialization)

---

References

1. https://docs.djangoproject.com/en/5.0/topics/async/ 2. https://www.django-rest-framework.org/api-guide/pagination/ 3. https://opensearch.org/docs/latest/search-plugins/searching-data/paginate/ 4. https://opensearch.org/docs/latest/query-dsl/compound/function-score/ 5. https://docs.aws.amazon.com/personalize/latest/dg/getting-real-time-recommendations.html 6. https://docs.databricks.com/en/machine-learning/model-serving/index.html 7. https://www.python-httpx.org/async/ 8. https://docs.python.org/3/library/asyncio-task.html 9. https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/ 10. https://redis.io/docs/manual/patterns/distributed-locks/ 11. https://datatracker.ietf.org/doc/html/rfc5861 12. https://github.com/danielfm/pybreaker 13. https://www.uvicorn.org/deployment/

---

Source Files

This document was compiled from individual reference files. For detailed editing or extension:

{Title}

{1-3 sentences explaining WHY this matters. What goes wrong without this pattern in a Django backend serving recommendations + search? Frame the failure in concrete terms: extra downstream load, blown SLO budget, cascading worker exhaustion, ML quota burn, cache miss storm, OpenSearch shard pressure. The mechanism is what makes the rule generalize to novel scenarios.}

Incorrect ({problem label}):

{Production-realistic bad code — use names like fetch_recommendations, search_products,
PersonalizeClient, opensearch_client, not foo/bar.}
{Comments explain the *cost*: "# blocks event loop" or "# fires N requests".}

async def bad_example():
    items = ...  # 🚨 explanation of what's wrong

Correct ({solution label}):

{Good code — minimal diff from incorrect when possible.}
{Comments explain the *benefit*.}

async def good_example():
    items = await with_circuit_breaker(
        lambda: personalize_client.get(...),
    )                                       # ← the fix

{Optional sections — include only when they add value:}

Alternative ({context}):

{Alternative valid approach}

Implementation ({name of pattern}):

{Reusable utility worth shipping with the rule}

With {framework/tool}:

{Tool-specific variant — e.g., boto3, httpx, opensearch-py, redis.asyncio}

When NOT to use this pattern:

• {Specific exception with rationale}
• {Another specific exception}

Warning ({context}):

• {Gotcha that would burn a careful reader}

Pair with [[other-rule-slug]]: {how this rule combines with another}

Reference: [{Source Title}]({source URL — use authoritative sources only: official docs, AWS blog, RFC, primary maintainers})

Sections

This file defines all sections, their ordering, impact levels, and descriptions. The section ID (in parentheses) is the filename prefix used to group rules.

Categories are ordered by lifecycle position and cascade effect. Problems at the top (fan-out orchestration, external service protection, search query design) multiply downstream — getting them wrong creates p99 spikes, cascading downstream failures, or wasted compute on every request. Problems at the bottom (response serialization, pagination shape) are localized.

Impact tier definitions

Used by rule frontmatter and category headings:

---

1. Fan-out Orchestration (orch)

Impact: CRITICAL Description: How concurrent calls to Personalize, internal microservices, and Databricks ML endpoints are coordinated — asyncio.gather with return_exceptions=True, deadline propagation across hops, partial-result aggregation, parallel-not-serial fan-out, async client reuse. Wrong here turns a 200ms p99 into a 2-second p99 (slowest-downstream bottleneck) or one downstream failure into a full request failure.

2. External Service Protection (protect)

Impact: CRITICAL Description: Per-downstream guardrails — circuit breakers tuned per service (Personalize fails differently than Databricks), per-endpoint timeout budgets, full-jitter exponential backoff, bulkhead pools, client-side rate-limiting toward downstreams. Without these, one slow service hangs every worker thread; one outage cascades into client-quota exhaustion.

3. OpenSearch Query Patterns (search)

Impact: CRITICAL Description: Index, query, and pagination design — search_after cursor instead of deep from/size, _source filtering to bound payload size, function_score for blending personalization signals, request cache enablement, index aliases for blue/green, tie-breaker sort for stable cursors. Bad query design turns a 50ms search into a 5-second search at production data volume.

4. Result Blending & Personalization (blend)

Impact: HIGH Description: Mixing heterogeneous recommender outputs — score normalization across sources with different score scales (Personalize 0..1, Databricks logits, OpenSearch BM25), MMR diversity to avoid recommendation monocultures, cross-source dedup by canonical item ID, cold-start fallback for new users, anonymous-vs-personalized response splits.

5. Caching Strategy (cache)

Impact: HIGH Description: Redis-tier patterns for an API serving expensive ML/search calls — key design with user-segment isolation, stampede protection via SETNX locks + jittered TTLs, cache versioning keyed on ML model deploy, two-tier caching (process LRU + Redis), negative caching for empty results. Wrong here means thundering-herd refetches on TTL expiry and stale recommendations after a model swap.

6. Resilience & Partial Results (resilience)

Impact: HIGH Description: How the API degrades when some downstreams fail — partial: true response envelopes flagging which sources contributed, stale-from-Redis fallback when fresh fetch fails, default ranking fallback when all recommenders are down, per-source observability tags so downstream failures are visible in metrics. Without these, one Databricks blip becomes a 500 for the entire recommendation page.

7. Async & Concurrency (async)

Impact: MEDIUM-HIGH Description: Django-specific async patterns — async views vs. sync, sync_to_async/async_to_sync correctness, ORM async pitfalls, gunicorn worker model vs. uvicorn for IO-bound workloads, connection pool sizing per downstream, httpx.AsyncClient reuse across requests. Wrong here either negates async benefits (blocking-in-async) or corrupts request-scoped state.

8. API Response Design (api)

Impact: MEDIUM Description: DRF-level response shape — cursor pagination instead of page-number, serializer performance (select_related/prefetch_related/only/values), ETag/Cache-Control headers for CDN/proxy reuse, streaming responses for large result sets, response compression. These don't change throughput dramatically but compound across endpoints.

Compress Responses and Shape Payloads

A 50KB JSON response compresses to ~6KB with gzip and ~5KB with brotli — that's bandwidth, CDN cost, and time-to-first-byte cut by 8-10×. For an API serving millions of requests/day, compression alone is one of the largest cost levers. Pair with payload shaping: drop optional fields, use shorter keys for hot paths, and consider binary formats (msgpack, protobuf) for high-volume internal traffic.

Compression is mostly free CPU-wise on modern servers (~1-5% overhead) and the bandwidth savings dwarf it. Make sure it's enabled at the right layer.

Incorrect (no compression — sending uncompressed JSON):

# settings.py — no compression middleware
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.contrib.sessions.middleware.SessionMiddleware",
    ...
]

# Response: 50KB sent on the wire for every request

Correct (enable Django's GZip middleware):

# settings.py — put GZipMiddleware near the top, AFTER security
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "django.middleware.gzip.GZipMiddleware",   # ✅ compress responses
    "django.contrib.sessions.middleware.SessionMiddleware",
    ...
]

# Response: 6KB sent on the wire (~10× smaller)
# Django auto-detects Accept-Encoding and only compresses when client supports it

Better: compress at the reverse proxy (nginx) — frees Django CPU:

# nginx.conf
gzip on;
gzip_vary on;
gzip_min_length 256;
gzip_types
    application/json
    application/javascript
    text/css text/plain text/xml
    application/xml application/xml+rss
    image/svg+xml;
gzip_comp_level 5;  # 1-9; higher = more CPU, more compression. 5-6 is the sweet spot.

# Or brotli (better compression, requires the brotli nginx module):
brotli on;
brotli_types application/json application/javascript text/css text/plain;
brotli_comp_level 5;
brotli_static on;

When the reverse proxy compresses, Django doesn't need its own middleware. Pick one — not both.

Shape payloads — drop fields the client doesn't render:

# ❌ Returning the full Product model in every list response
class ProductSerializer(ModelSerializer):
    class Meta:
        model = Product
        fields = "__all__"  # 30 fields, even when the UI shows 5

# ✅ Shape per use case
class ProductListSerializer(serializers.Serializer):
    id = serializers.IntegerField()
    title = serializers.CharField()
    price = serializers.DecimalField(max_digits=10, decimal_places=2)
    thumbnail_url = serializers.URLField()
    in_stock = serializers.BooleanField()
# 5 fields × N items, 6× smaller than __all__

Conditional fields via `fields` query param (sparse fieldsets):

class FlexibleSerializer(serializers.ModelSerializer):
    """Client passes ?fields=id,title to limit serialization."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        request = self.context.get("request")
        if request is None:
            return
        fields_param = request.query_params.get("fields")
        if fields_param:
            allowed = set(fields_param.split(","))
            existing = set(self.fields)
            for field in existing - allowed:
                self.fields.pop(field)

class ProductSerializer(FlexibleSerializer):
    class Meta:
        model = Product
        fields = "__all__"

# Client requests:
# GET /products?fields=id,title,price
# Response: only id, title, price for each — ~50% smaller

For high-volume internal APIs, consider msgpack:

import msgpack

class MsgPackRenderer(renderers.BaseRenderer):
    media_type = "application/msgpack"
    format = "msgpack"

    def render(self, data, accepted_media_type=None, renderer_context=None):
        return msgpack.packb(data, use_bin_type=True)

class MyView(APIView):
    renderer_classes = [JSONRenderer, MsgPackRenderer]

# Internal consumers request: Accept: application/msgpack
# Response is binary — typically 30-50% smaller than JSON before compression

Drop nulls — JSON nulls are visual noise and bytes:

class CompactSerializer(serializers.Serializer):
    def to_representation(self, instance):
        data = super().to_representation(instance)
        return {k: v for k, v in data.items() if v is not None}
# {"id": 1, "title": "...", "discount": null} → {"id": 1, "title": "..."}

Be careful with this — some clients expect explicit null to distinguish "missing from response" from "present but null."

Use shorter keys for high-volume responses:

# ❌ Verbose keys multiplied across thousands of items
{
    "product_identifier": 42,
    "product_display_title": "...",
    "current_market_price": 19.99,
}

# ✅ For internal feeds where size matters more than readability
{
    "id": 42,
    "title": "...",
    "price": 19.99,
}

For public APIs, prefer readability. For internal high-volume APIs (mobile app payloads), every byte counts.

Pre-compress static-ish responses (e.g., catalog data):

# Compute and cache the gzipped response in Redis
import gzip

async def get_compressed_catalog(segment: str) -> bytes:
    cached = await redis.get(f"catalog:gzip:{segment}")
    if cached:
        return cached

    items = await get_catalog(segment)
    raw = json.dumps(items).encode()
    compressed = gzip.compress(raw, compresslevel=6)
    await redis.setex(f"catalog:gzip:{segment}", 3600, compressed)
    return compressed

# Return pre-compressed bytes directly
async def catalog_view(request):
    body = await get_compressed_catalog(_segment(request))
    response = HttpResponse(body, content_type="application/json")
    response["Content-Encoding"] = "gzip"
    response["Vary"] = "Accept-Encoding"
    return response

Verify compression is working:

curl -i -H "Accept-Encoding: gzip, deflate, br" https://api.example.com/recommendations
# Look for: Content-Encoding: gzip (or br)
# Look at: Content-Length — should be much smaller than the uncompressed size

Don't compress already-compressed content:

Images, videos, gzipped archives. Compressing them again costs CPU for ~0 benefit. Nginx's gzip_types directive restricts to text formats by default — good.

Symptom of missing compression:

• Egress bandwidth costs disproportionate to RPS
• Mobile users complain about data usage
• TTFB high on responses > 10KB

Reference: Django — GZipMiddleware | nginx — gzip module | msgpack

Return Cursor-Based Pagination from DRF

DRF's default PageNumberPagination (?page=2&size=20) returns the same items shifted around as data changes between requests — the user sees duplicates and missing items on infinite scroll. CursorPagination solves this by encoding the position with the last item's sort key, returning an opaque cursor the client passes back.

For recommendation feeds, search results, and any list that changes between page fetches (which is most lists), use cursor pagination. The response shape is stable, future-proof, and aligns with how the OpenSearch backend already paginates (search_after, see [[search-use-search-after-not-from]]).

Incorrect (page-number pagination with shifting data):

# views.py
from rest_framework.pagination import PageNumberPagination

class StandardPageNumberPagination(PageNumberPagination):
    page_size = 20
    max_page_size = 100

class RecommendationsList(ListAPIView):
    pagination_class = StandardPageNumberPagination
# Response:
# {
#   "count": 1037,
#   "next": "https://api.example.com/recommendations?page=3",
#   "previous": "...page=1",
#   "results": [...]
# }
# Problem: new items inserted between page 1 and page 2 → user sees duplicates

Correct (cursor pagination — stable across writes):

from rest_framework.pagination import CursorPagination

class RecommendationsCursorPagination(CursorPagination):
    page_size = 20
    max_page_size = 100
    ordering = "-created_at"     # tiebreaker — see [[search-stable-tiebreaker-sort]]
    cursor_query_param = "cursor"

class RecommendationsList(ListAPIView):
    pagination_class = RecommendationsCursorPagination
    queryset = Recommendation.objects.all()
# Response:
# {
#   "next": "https://api.example.com/recommendations?cursor=cD0yMDI2LTA1LTE5",
#   "previous": null,
#   "results": [...]
# }
# Opaque cursor; stable across inserts; no skipping/duplicating

For OpenSearch-backed endpoints (search_after under the hood):

DRF's CursorPagination is for queryset-based endpoints. For OpenSearch-backed endpoints, build a custom paginator that wraps search_after:

import base64
import json
from rest_framework.pagination import BasePagination
from rest_framework.response import Response

class OpenSearchCursorPagination(BasePagination):
    """Opaque cursor pagination backed by OpenSearch search_after."""
    page_size = 20
    max_page_size = 100

    def paginate(self, query_fn, request, view=None):
        """Call this from your view; query_fn accepts (cursor, size) → (items, next_cursor_sort_values)."""
        size = self._get_size(request)
        cursor = self._decode(request.query_params.get("cursor"))
        items, next_sort = query_fn(cursor=cursor, size=size)
        self._next_cursor = self._encode(next_sort) if next_sort else None
        self._items = items
        self._request = request
        return items

    def get_paginated_response(self, data):
        next_url = self._build_next_url() if self._next_cursor else None
        return Response({
            "items": data,
            "next": next_url,
            "page_size": len(data),
        })

    def _get_size(self, request) -> int:
        size = request.query_params.get("size")
        try:
            size = int(size) if size else self.page_size
        except ValueError:
            size = self.page_size
        return min(max(1, size), self.max_page_size)

    def _encode(self, sort_values) -> str:
        return base64.urlsafe_b64encode(
            json.dumps(sort_values).encode()
        ).decode().rstrip("=")

    def _decode(self, cursor: str | None):
        if not cursor:
            return None
        try:
            padded = cursor + "=" * (-len(cursor) % 4)
            return json.loads(base64.urlsafe_b64decode(padded))
        except (ValueError, json.JSONDecodeError):
            return None  # malformed cursor → start over

    def _build_next_url(self) -> str:
        from urllib.parse import urlencode, urlparse, urlunparse, parse_qs
        parts = urlparse(self._request.build_absolute_uri())
        qs = parse_qs(parts.query)
        qs["cursor"] = [self._next_cursor]
        return urlunparse(parts._replace(query=urlencode(qs, doseq=True)))

Usage in the view:

class SearchView(APIView):
    pagination_class = OpenSearchCursorPagination

    def get(self, request):
        paginator = self.pagination_class()
        items = paginator.paginate(
            query_fn=lambda cursor, size: search_opensearch(
                query=request.query_params["q"], cursor=cursor, size=size,
            ),
            request=request, view=self,
        )
        return paginator.get_paginated_response([self._serialize(i) for i in items])

Response shape consistency across endpoints:

Use the same envelope across the API so clients don't branch:

# Standard response shape for all paginated endpoints
{
    "items": [...],
    "next": "...?cursor=abc",   # or null on last page
    "page_size": 20,
}

# For homepage feeds with multiple sections
{
    "sections": [
        {"name": "for_you", "items": [...], "next": "...?cursor=..."},
        {"name": "trending", "items": [...], "next": "...?cursor=..."},
    ]
}

Don't expose total counts on cursor-paginated endpoints:

Counting all matches is expensive (OpenSearch needs to count every shard; SQL needs a full count query). Cursor pagination doesn't need totals — the existence of next tells the client there's more. Drop the count field from response.

If users really need a total ("Showing X of Y results"), make it a separate endpoint that's cached aggressively or returns an approximate count (total_approximate: true).

Encode cursors opaquely:

# ❌ Exposing sort fields lets clients fabricate cursors and skip security checks
?cursor=created_at:2026-05-19,id:abc123

# ✅ Opaque base64-encoded JSON — clients can only echo it back
?cursor=eyJzb3J0IjpbIjIwMjYtMDUtMTkiLCJhYmMxMjMiXX0

Sign cursors if they carry sensitive sort values:

import hmac, hashlib

def sign_cursor(sort_values, secret: bytes) -> str:
    payload = json.dumps(sort_values).encode()
    sig = hmac.new(secret, payload, hashlib.sha256).digest()[:8]
    return base64.urlsafe_b64encode(payload + b"|" + sig).decode().rstrip("=")

This prevents clients from constructing arbitrary cursors to scan data they shouldn't see.

Symptom of bad pagination:

• "User reports seeing the same item on consecutive pages" — page-number pagination + writes
• "Search page 100 returns 500" — from + size > 10000
• "Total count query takes 5 seconds" — exposing count on every page

Reference: DRF — Pagination | Use The Index, Luke — Pagination

Set ETag and Cache-Control for CDN/Client Reuse

A response without Cache-Control headers can't be cached by CDNs or browsers — every request hits Django. A response with ETag lets clients re-validate (sending If-None-Match) and get a 304 Not Modified (no body) instead of the full response. For an API serving largely-stable data (popular feeds, search-result snapshots), these headers cut Django load to <10% of the otherwise-needed capacity.

The trick is correctness: don't set long TTLs on per-user data, don't cache partial responses ([[resilience-partial-response-envelope]]) as long as fresh ones, and use the Vary header to prevent cross-segment serving.

Incorrect (no caching headers — every request hits Django):

def popular_view(request):
    items = get_popular_items()
    return JsonResponse({"items": items})
# CDN: can't cache, sends every request to origin
# Browser: must re-fetch every page load

Correct (ETag + Cache-Control + Vary):

import hashlib
import json

def popular_view(request):
    items = get_popular_items(segment=_segment(request))

    body = {"items": items, "generated_at": ...}
    body_bytes = json.dumps(body, sort_keys=True).encode()

    etag = '"' + hashlib.blake2b(body_bytes, digest_size=8).hexdigest() + '"'

    # Conditional GET — if the client sent If-None-Match, compare
    if request.META.get("HTTP_IF_NONE_MATCH") == etag:
        return HttpResponse(status=304)  # no body needed; client uses its cache

    response = HttpResponse(body_bytes, content_type="application/json", status=200)
    response["ETag"] = etag
    response["Cache-Control"] = "public, max-age=60, stale-while-revalidate=300"
    response["Vary"] = "Accept-Language, Authorization"
    return response

Cache-Control directives (composition matters):

Common combinations:

Vary header — prevent cross-context serving:

The Vary header tells CDNs which request headers affect the response. Without it, CDN may return the English response to an Italian user, or an authenticated response to an anonymous user:

response["Vary"] = "Accept-Language, Authorization, X-Tenant-Id"

Common Vary values:

• Accept-Language — localized responses
• Authorization — anonymous vs authenticated
• Cookie — when cookies affect the response (be careful: cache miss rate explodes)
• Accept-Encoding — for compressed responses (most servers add this automatically)
• X-*-Id — for tenant or segment headers

Don't include `Cookie` in Vary unless necessary:

Vary: Cookie means every distinct cookie value gets its own cache entry. Since cookies are typically per-user, this defeats CDN caching entirely. If only certain cookies affect the response, use a more specific header (e.g., a X-User-Segment header set by your edge logic).

Generate ETags from content hash, not random:

# Hash the response body (deterministic)
etag = '"' + hashlib.blake2b(body_bytes, digest_size=8).hexdigest() + '"'

# Or from a version + timestamp
etag = f'"v23-{int(updated_at.timestamp())}"'

# Or weak ETag (less strict comparison)
etag = f'W/"abc123"'  # weak — semantic equivalence, not byte-for-byte

Cache the ETag itself for expensive computations:

async def search_view(request):
    cache_key = f"search:{hash(request.GET['q'])}:etag"
    cached_etag = await redis.get(cache_key)

    if cached_etag and request.META.get("HTTP_IF_NONE_MATCH") == cached_etag.decode():
        return HttpResponse(status=304)

    items = await opensearch_search(request.GET["q"])
    body = {"items": items}
    body_bytes = json.dumps(body, sort_keys=True).encode()
    etag = '"' + hashlib.blake2b(body_bytes, digest_size=8).hexdigest() + '"'

    await redis.setex(cache_key, 300, etag)

    response = HttpResponse(body_bytes, status=200)
    response["ETag"] = etag
    response["Cache-Control"] = "public, max-age=60, stale-while-revalidate=300"
    return response

Don't cache responses with auth-dependent data without `Vary: Authorization`:

# ❌ CDN may serve User A's account data to User B
response["Cache-Control"] = "public, max-age=300"
# (no Vary)

# ✅ Either private OR vary by auth
response["Cache-Control"] = "private, max-age=300"   # user-only cache
# or
response["Cache-Control"] = "public, max-age=300"
response["Vary"] = "Authorization"                    # CDN keys by auth header

Use 304 to save bandwidth, not 200 with same body:

# ❌ Sending the same response with same ETag every time wastes bandwidth
# ✅ Return 304 when If-None-Match matches
if request.META.get("HTTP_IF_NONE_MATCH") == etag:
    return HttpResponse(status=304)  # no body — client uses its cached copy

Symptom of missing cache headers:

• "Why is the API hit so much for popular endpoints?" — no CDN cacheability
• CDN hit ratio < 30% on cacheable endpoints
• Mobile users complain about data usage for repeated visits

Reference: MDN — Cache-Control | MDN — ETag | Django — HTTP shortcut decorators

Use select_related, prefetch_related, and only in DRF Serializers

A DRF ModelSerializer with a nested UserSerializer field will issue one query per row to fetch the user — the N+1 problem. For a 20-item recommendations response, that's 21 database queries instead of 1. DRF doesn't auto-detect related fields you serialize; you have to specify what to join via select_related/prefetch_related on the queryset.

only(...) further bounds the work: by default, every column is fetched. For a Product with 30 columns where the serializer renders 5, only("id", "title", "price", "thumbnail_url", "in_stock") cuts row size 6× and eliminates expensive column fetches (e.g., serialized JSON, large text fields).

Incorrect (N+1 serialization, full-row fetch):

# serializers.py
class RecommendationSerializer(serializers.ModelSerializer):
    product = ProductSerializer()    # nested
    user = UserSerializer()           # nested

    class Meta:
        model = Recommendation
        fields = ["id", "user", "product", "score", "created_at"]

# views.py
class RecommendationsList(ListAPIView):
    serializer_class = RecommendationSerializer
    queryset = Recommendation.objects.all()   # ❌ no joins, full columns

# For 20 results:
#   1 query: SELECT * FROM recommendations LIMIT 20
#  20 queries: SELECT * FROM products WHERE id = ?  (per row!)
#  20 queries: SELECT * FROM users WHERE id = ?
# Total: 41 queries

Correct (eager joins + column projection):

class RecommendationsList(ListAPIView):
    serializer_class = RecommendationSerializer

    def get_queryset(self):
        return (
            Recommendation.objects
            .select_related("product", "user")           # ✅ join in 1 query
            .only(
                "id", "score", "created_at",
                "product__id", "product__title", "product__price", "product__thumbnail_url",
                "user__id", "user__name", "user__avatar_url",
            )                                             # ✅ project only needed columns
        )

# 1 query total: SELECT recommendations.*, products.*, users.* FROM ... JOIN ... JOIN ...
# 4× faster, smaller transfer, less DB load

select_related vs prefetch_related:

# For many-to-one (one product belongs to one category):
qs = Product.objects.select_related("category")

# For many-to-many (one product has many tags):
qs = Product.objects.prefetch_related("tags")

# Combined:
qs = (
    Product.objects
    .select_related("category", "brand")           # 1:1 / many:1 → joins
    .prefetch_related("tags", "variants")          # 1:many / m:m → in-clauses
    .only("id", "title", "price", "category__name", "brand__name")
)

For reverse relations with constraints, use Prefetch:

from django.db.models import Prefetch

# Fetch only active variants for each product
qs = Product.objects.prefetch_related(
    Prefetch(
        "variants",
        queryset=Variant.objects.filter(is_active=True).only("id", "name", "stock"),
        to_attr="active_variants",
    )
)
# Each product now has product.active_variants
# Without the to_attr, it would shadow product.variants (the manager)

Use `values_list` for trivial responses (skip serializer overhead):

For endpoints that just return a list of IDs or simple tuples:

# Slow: full ORM + serializer
ids = list(Product.objects.filter(...).values("id"))
# Fast: direct values_list (no model instantiation)
ids = list(Product.objects.filter(...).values_list("id", flat=True))
# Faster still on huge lists: queryset.iterator(chunk_size=1000)

Detect N+1 in development:

# requirements-dev.txt
django-debug-toolbar

# settings.py (dev only)
INSTALLED_APPS += ["debug_toolbar"]
MIDDLEWARE += ["debug_toolbar.middleware.DebugToolbarMiddleware"]

# Or use nplusone for automated detection
INSTALLED_APPS += ["nplusone.ext.django"]
MIDDLEWARE += ["nplusone.ext.django.NPlusOneMiddleware"]
NPLUSONE_RAISE = True   # fails tests on N+1

nplusone raises an exception on every detected N+1, surfacing them in CI before they reach production.

Avoid `to_representation` overrides that re-query:

# ❌ Subtle N+1: to_representation calls .first() on a related queryset
class ProductSerializer(serializers.ModelSerializer):
    primary_image = serializers.SerializerMethodField()

    def get_primary_image(self, obj):
        return obj.images.first().url   # ← one query per product

# ✅ Use Prefetch + first item in Python
class ProductSerializer(serializers.ModelSerializer):
    primary_image = serializers.SerializerMethodField()

    def get_primary_image(self, obj):
        # Relies on Prefetch having already loaded obj.images
        images = list(obj.images.all())  # uses prefetched data
        return images[0].url if images else None

For large nested responses, consider flat serialization:

When the nested object is heavy (lots of fields, deep nesting), flatten it:

# Heavy nested
{"user": {"id": ..., "name": ..., "email": ..., "preferences": {...}}}

# Flat with just the fields needed
{"user_id": ..., "user_name": ...}

# Implement with SerializerMethodField:
class RecommendationSerializer(serializers.Serializer):
    user_id = serializers.IntegerField(source="user.id")
    user_name = serializers.CharField(source="user.name")
    # ... no nested UserSerializer

Symptom of N+1 in serialization:

• Endpoint latency scales linearly with page_size
• Database query count > 5 per simple list endpoint
• "Single row fast, list slow" mismatch

Reference: Django — Database access optimization | DRF — Optimizing serialization | nplusone

Apply Throttling per User and per Expensive Endpoint

A logged-in user that issues 200 recommendation requests per second (intentional scraper, broken client, or test code left running) can exhaust your Personalize/Databricks quota for everyone else. Without throttling, one misbehaving caller becomes everyone's outage. DRF's Throttle classes apply per-user and per-endpoint limits at the Django layer — cheaper than letting requests flow through to expensive downstreams.

Throttling is the user-facing complement to [[protect-client-side-rate-limit]] (which throttles outbound to downstreams). Use both: API throttling bounds which users can call you frequently; outbound throttling bounds how often you call downstreams.

Incorrect (no throttling — one user can drain quota for everyone):

class RecommendationsView(APIView):
    def get(self, request):
        items = expensive_recommendation_call(request.user.id)
        return Response({"items": items})
# A bot can hit this 1000 RPS per user; each call burns Personalize quota

Correct (per-user throttle + per-endpoint anonymous throttle):

# settings.py
REST_FRAMEWORK = {
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.UserRateThrottle",
        "rest_framework.throttling.AnonRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "user": "1000/hour",                # default — broad cap
        "anon": "100/hour",                  # anonymous traffic — tighter
        "recommendations_user": "60/minute", # specific expensive endpoint
        "recommendations_anon": "20/minute",
        "search_user": "120/minute",
        "search_anon": "30/minute",
    },
}

# views.py
from rest_framework.throttling import UserRateThrottle, AnonRateThrottle

class RecommendationsUserThrottle(UserRateThrottle):
    scope = "recommendations_user"

class RecommendationsAnonThrottle(AnonRateThrottle):
    scope = "recommendations_anon"

class RecommendationsView(APIView):
    throttle_classes = [RecommendationsUserThrottle, RecommendationsAnonThrottle]

    def get(self, request):
        items = expensive_recommendation_call(request.user.id)
        return Response({"items": items})

Throttle by tier — paying customers get higher limits:

class TieredUserThrottle(UserRateThrottle):
    """Higher rate limit for paying users."""

    def get_rate(self):
        request = self._request_context
        if request and request.user.is_authenticated:
            if request.user.is_premium:
                return "300/minute"
            if request.user.is_paid:
                return "120/minute"
        return "60/minute"   # free tier

Throttle by IP for anonymous (prevent single-IP abuse):

class IpThrottle(AnonRateThrottle):
    """Per-IP throttle for anonymous endpoints — different from per-anon-session."""
    scope = "ip"

    def get_ident(self, request):
        # Use the trusted forwarded IP (depends on your proxy setup)
        return request.META.get("HTTP_X_FORWARDED_FOR", "").split(",")[0].strip() \
            or request.META.get("REMOTE_ADDR")

For burst-tolerant traffic, use a token bucket (not DRF's simple bucket):

DRF's throttle is a sliding window — once the bucket is full, the limit is hard. For traffic that has bursts but stays within average, a token bucket is more permissive:

# Custom DRF throttle backed by Redis token bucket
class TokenBucketThrottle:
    rate = 60       # tokens per minute (sustained rate)
    burst = 20      # extra tokens for short bursts

    def allow_request(self, request, view):
        user_id = request.user.id if request.user.is_authenticated else \
                  request.META.get("REMOTE_ADDR")
        key = f"throttle:{view.__class__.__name__}:{user_id}"
        return redis_token_bucket_acquire(key, self.rate, self.burst)

Throttle by request cost, not just count:

Some endpoints are 10× more expensive than others (deep search, full personalization). Charge multiple "tokens" per expensive call:

class CostBasedThrottle(BaseThrottle):
    def allow_request(self, request, view):
        cost = getattr(view, "throttle_cost", 1)
        return redis_token_bucket_charge(self._key(request), cost)

class DeepSearchView(APIView):
    throttle_classes = [CostBasedThrottle]
    throttle_cost = 5    # this view costs 5 tokens per call

class CheapEndpointView(APIView):
    throttle_classes = [CostBasedThrottle]
    throttle_cost = 1

Return Retry-After on 429s — clients can honor it ([[protect-honor-retry-after-header]]):

# DRF does this automatically when throttle.wait() returns a value
class RecommendationsUserThrottle(UserRateThrottle):
    scope = "recommendations_user"
    # DRF automatically sets:
    #   429 Too Many Requests
    #   Retry-After: <seconds>

Bypass throttling for internal traffic (with care):

class ExternalOnlyThrottle(UserRateThrottle):
    """Skip throttle for requests from internal services (e.g., via a shared secret)."""

    def allow_request(self, request, view):
        if request.META.get("HTTP_X_INTERNAL_API_KEY") == settings.INTERNAL_API_KEY:
            return True
        return super().allow_request(request, view)

Don't throttle on health-check / status endpoints:

class HealthView(APIView):
    throttle_classes = []  # health checks must always work
    permission_classes = []  # no auth either

    def get(self, request):
        return Response({"status": "ok"})

Throttled health checks make load balancers mark instances unhealthy on rate-limit spikes — wrong signal.

Observability — track throttle hits:

class ObservableThrottle(UserRateThrottle):
    def throttle_failure(self):
        metrics.increment("throttle.exceeded", tags={
            "scope": self.scope,
            "view": getattr(self.view, "__class__", None).__name__,
        })
        return super().throttle_failure()

A spike in throttle.exceeded for one scope often surfaces abuse or a broken client.

Don't throttle so aggressively that legitimate clients break:

Pick rates by measuring actual traffic from your highest-volume legitimate users (analytics service, mobile app, automated tests) and setting limits well above their natural rate. Throttling should catch outliers, not bound average traffic.

Symptom of missing throttle:

• One user/IP causes API-wide latency spikes
• Personalize/Databricks bills correlate with traffic from a single source
• Abuse complaints from competitors trying to scrape

Reference: DRF — Throttling | django-ratelimit | Stripe — Rate Limiters

Cancel In-Flight Work When the Client Disconnects

A user navigates away mid-request. Their browser closes the connection. The Django worker is still happily calling Personalize + Databricks + OpenSearch + blending — burning ML inference cost, database connections, and downstream rate-limit quota on a response no one will see. Worse, the worker remains busy until everything completes, so the next request waits.

Under ASGI, Django can detect the disconnect and cancel the request. The pattern: check await request.is_disconnected() periodically, or rely on asyncio.CancelledError propagation. For downstream calls, propagate cancellation so AWS Personalize/Databricks stop processing too (where supported).

Incorrect (no disconnect detection — work continues after client gone):

async def recommendations_view(request):
    # User closes browser after 100ms. The view doesn't know.
    user_id = request.user.id
    personalize, affinity, databricks = await asyncio.gather(
        personalize_client.get(user_id),    # still running
        affinity_client.get(user_id),       # still running
        databricks_client.invoke(user_id),  # still running
    )
    items = blend_results([personalize, affinity, databricks])
    return JsonResponse(items)
    # Worker holds resources for the full duration. Wasted.

Correct (check disconnect; propagate cancellation):

async def recommendations_view(request):
    user_id = request.user.id

    # Wrap the fan-out so we can race it against disconnect detection
    work_task = asyncio.create_task(_fanout_and_blend(user_id))
    disconnect_task = asyncio.create_task(_wait_for_disconnect(request))

    done, pending = await asyncio.wait(
        [work_task, disconnect_task],
        return_when=asyncio.FIRST_COMPLETED,
    )

    if disconnect_task in done:
        # Client gone — cancel the work
        work_task.cancel()
        try:
            await work_task
        except (asyncio.CancelledError, BaseException):
            pass
        # No response needed; the connection is closed
        return HttpResponse(status=499)  # 499 = client closed request (nginx convention)

    # Work completed first — cancel the disconnect watcher
    disconnect_task.cancel()
    items = await work_task
    return JsonResponse({"items": items})

async def _wait_for_disconnect(request):
    while True:
        if await request.is_disconnected():
            return
        await asyncio.sleep(0.1)  # poll every 100ms

async def _fanout_and_blend(user_id: str):
    results = await asyncio.gather(
        personalize_client.get(user_id),
        affinity_client.get(user_id),
        databricks_client.invoke(user_id),
        return_exceptions=True,
    )
    return blend_results(results)

Propagate cancellation to HTTP clients:

httpx.AsyncClient honors asyncio.CancelledError — when the parent task is cancelled, in-flight requests are aborted. The downstream server may or may not stop work on disconnect (depends on its own implementation), but at least your worker frees up immediately:

async def _fanout_and_blend(user_id: str):
    # When this task is cancelled, httpx aborts the in-flight requests
    return await asyncio.gather(
        personalize_client.get(user_id),
        databricks_client.invoke(user_id),
        return_exceptions=True,
    )

For boto3 (sync, no async cancellation):

# boto3 doesn't honor cancellation. asyncio.to_thread propagates cancellation to the
# Python task but the boto3 call continues until completion.
# Best effort: use a short timeout in the boto3 client config so it can't run forever.

import boto3
from botocore.config import Config

_personalize = boto3.client("personalize-runtime", config=Config(
    connect_timeout=1.0,
    read_timeout=2.0,    # caps the worst case
))

Use ASGI lifespan for graceful shutdown:

# settings/asgi.py
import asyncio
from django.core.asgi import get_asgi_application
from django.conf import settings

django_application = get_asgi_application()

async def application(scope, receive, send):
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                # Cancel background tasks here
                await _shutdown_background_tasks()
                await send({"type": "lifespan.shutdown.complete"})
                return
    else:
        await django_application(scope, receive, send)

Don't over-engineer for low-stakes endpoints:

For most recommendation endpoints, the full work is <1s. The user's window for disconnecting is small. Implementing disconnect detection is worth it only when:

• The work is expensive enough to matter (ML inference calls, large query fan-outs)
• p95 latency is high enough that abandonment is common (>500ms)
• You're hitting rate limits or quota constraints

For sub-100ms endpoints, skip the complexity.

Tradeoff with caching:

Cancelling a downstream call means the cache won't be populated for the next user. Sometimes you want the call to complete even after the original user disconnected, because the next user is about to ask the same thing. For shared-cache endpoints (popular feed, common search), let the request finish:

async def search_view(request):
    if _is_shared_cache_endpoint(request):
        # Don't cancel — let the work warm the cache for others
        return await _search(request)
    # User-specific endpoint — cancel on disconnect
    return await _search_cancellable(request)

Don't cancel partial database writes (data integrity):

If your view writes to the DB then returns, cancellation between write and response means the data is committed but the user thinks the request failed. They retry, you double-write. Either avoid cancellation for write paths, or use transactions + idempotency keys.

Symptom of missing cancellation:

• Worker pool exhaustion after a traffic burst with high abandon rate
• "Personalize bill is higher than expected" — wasted calls on abandoned requests
• p99 latency dragged up by old requests still processing

Reference: Django — Async views and request handling | Python — asyncio cancellation

Use contextvars for Request-Scoped State Across Async Calls

In sync Django, request-scoped state (the current user, request ID, locale) often lives in thread-local storage (threading.local()). This works because each request gets a dedicated thread. In async views, one thread handles many concurrent requests — thread-local storage spills across requests, and a value set by request A is visible to request B running on the same thread between awaits.

contextvars.ContextVar is the asyncio-aware equivalent: variables are bound to the current task / async context, so each request has its own logical "thread-local" storage even when threads are shared. Use this for request IDs, current user, deadlines, locale — anything that needs to travel through nested async calls without explicit parameter threading.

Incorrect (thread-local in async — cross-request contamination):

# auth.py
import threading
_thread_local = threading.local()

def set_current_user(user):
    _thread_local.user = user  # ❌ shared across async requests

def get_current_user():
    return getattr(_thread_local, "user", None)

# middleware.py
async def auth_middleware(get_response, request):
    set_current_user(request.user)
    response = await get_response(request)  # ← awaits here; other requests run on same thread
    return response

# Inside another concurrent request, get_current_user() returns wrong user

Correct (contextvars — isolated per async task):

# auth.py
import contextvars

_current_user: contextvars.ContextVar = contextvars.ContextVar("current_user", default=None)

def set_current_user(user):
    return _current_user.set(user)  # returns a Token for restoring

def get_current_user():
    return _current_user.get()

def reset_current_user(token):
    _current_user.reset(token)

# middleware.py
async def auth_middleware(get_response, request):
    token = set_current_user(request.user)
    try:
        response = await get_response(request)
    finally:
        reset_current_user(token)
    return response
# Each request has its own value; no leakage across concurrent requests

Common request-scoped contextvars:

# context.py
import contextvars
import time

request_id: contextvars.ContextVar[str | None] = contextvars.ContextVar("request_id", default=None)
current_user: contextvars.ContextVar = contextvars.ContextVar("current_user", default=None)
request_deadline: contextvars.ContextVar[float | None] = contextvars.ContextVar(
    "request_deadline", default=None
)
locale: contextvars.ContextVar[str] = contextvars.ContextVar("locale", default="en-US")

# Set them in middleware
async def context_middleware(get_response, request):
    tokens = [
        request_id.set(request.headers.get("X-Request-ID", str(uuid.uuid4()))),
        current_user.set(request.user),
        request_deadline.set(time.monotonic() + 0.5),
        locale.set(request.META.get("HTTP_ACCEPT_LANGUAGE", "en-US").split(",")[0]),
    ]
    try:
        return await get_response(request)
    finally:
        for t in tokens:
            try:
                t.var.reset(t)
            except LookupError:
                pass

Use them in deeply nested code (no parameter threading):

# clients/personalize.py — needs the request deadline but isn't passed it explicitly
async def get_recommendations(user_id: str):
    deadline = request_deadline.get()
    timeout = max(0.01, deadline - time.monotonic()) if deadline else 2.0

    return await client.post(url, json={...}, timeout=timeout)

Pass context to fire-and-forget tasks (or it's lost):

# By default, asyncio.create_task inherits the current context
async def search_view(request):
    # Context is set by middleware
    asyncio.create_task(track_event("search"))  # ✅ inherits context
    return JsonResponse({...})

async def track_event(event):
    user = current_user.get()        # works — inherited from request context
    req_id = request_id.get()
    await analytics.send(event, user_id=user.id, request_id=req_id)

Don't pass context to long-lived background workers (different lifecycle):

# Celery worker doesn't share the request's contextvars
@shared_task
def track_event_celery(event, user_id, request_id):
    # Pass values explicitly — they were captured at task submission time
    analytics.send(event, user_id=user_id, request_id=request_id)

# When scheduling, capture context explicitly:
track_event_celery.delay(
    event="search",
    user_id=current_user.get().id,
    request_id=request_id.get(),
)

For OpenTelemetry-style propagation:

Tracing libraries (OpenTelemetry, Datadog APM) already use contextvars under the hood for trace IDs. The pattern above is the same — just don't reinvent it for trace propagation; use the tracing library's API.

Don't store mutable state in contextvars:

# ❌ Storing a dict and mutating it — surprises on concurrent updates
request_state: contextvars.ContextVar[dict] = contextvars.ContextVar("state", default={})
request_state.get()["count"] += 1  # mutates the shared default dict across requests!

# ✅ Set a new dict each time
state = dict(request_state.get())
state["count"] = state.get("count", 0) + 1
request_state.set(state)

The default value is shared across all reads until .set() is called.

Thread-aware context bridging (when calling sync code from async):

# sync_to_async with thread_sensitive=True respects contextvars correctly
# (provided you've set them via contextvars, not threading.local)
result = await sync_to_async(sync_function, thread_sensitive=True)(args)
# sync_function can call current_user.get() and see the right value

Symptom of missing contextvars (using thread-locals in async):

• Random "wrong user" or "wrong request ID" in logs
• Sporadic "cross-request data leakage" reported in security review
• Tests pass locally (one request at a time) but fail under concurrent load

Reference: Python — contextvars | PEP 567 — Context Variables