Function stream_cached_response

apps/output/streaming_chunk_cache.py:190–239

Stream a large response with single-flight Redis chunk caching. ``source`` must be a callable returning a chunk iterator. Only the leader invokes it; concurrent followers replay chunks already written to Redis, so the expensive ``source`` runs at most once per ``cache_key``.

(
    cache_key,
    source,
    *,
    content_type="application/xml",
    filename=None,
    cache_ttl=DEFAULT_CACHE_TTL,
    lock_ttl=DEFAULT_LOCK_TTL,
    poll_interval=DEFAULT_POLL_INTERVAL,
    max_follower_wait=DEFAULT_MAX_FOLLOWER_WAIT,
    redis=None,
)

def stream_cached_response(
    cache_key,
    source,
    *,
    content_type="application/xml",
    filename=None,
    cache_ttl=DEFAULT_CACHE_TTL,
    lock_ttl=DEFAULT_LOCK_TTL,
    poll_interval=DEFAULT_POLL_INTERVAL,
    max_follower_wait=DEFAULT_MAX_FOLLOWER_WAIT,
    redis=None,
):
    """
    Stream a large response with single-flight Redis chunk caching.

    ``source`` must be a callable returning a chunk iterator. Only the leader
    invokes it; concurrent followers replay chunks already written to Redis, so
    the expensive ``source`` runs at most once per ``cache_key``.
    """
    if redis is None:
        redis = _get_redis()

    if redis.get(_ready_key(cache_key)):
        logger.debug("Serving response from chunk cache")
        stream = _stream_ready(redis, cache_key)
    else:
        status = _get_status(redis, cache_key)
        if status == STATUS_ERROR:
            _clear_build_keys(redis, cache_key)

        if _try_acquire_lock(redis, cache_key, lock_ttl):
            logger.debug("Building response (cache leader)")
            stream = _stream_build(redis, cache_key, source, cache_ttl, lock_ttl)
        else:
            logger.debug("Following in-flight cache build")
            stream = _stream_follow(
                redis,
                cache_key,
                source,
                cache_ttl,
                lock_ttl,
                poll_interval,
                max_follower_wait,
            )

    response = StreamingHttpResponse(stream, content_type=content_type)
    if filename:
        response["Content-Disposition"] = f'attachment; filename="{filename}"'
    response["Cache-Control"] = "no-cache"
    return response