AsyncResponse

Base class containing the core async content streaming logic.

Streaming retains no payload history: library memory stays proportional to the chunk size, and each send is awaited before the next read. All response classes accept an integer chunk_size from 1 through 1,048,576 bytes (1 MiB), default 4096. Booleans and other values raise ValueError at construction, before opening a stream or sending headers.

AsyncResponse uses Django's public StreamingHttpResponse interface: is_async is true and streaming_content is an async iterator. Middleware such as Django's GZipMiddleware can wrap that iterator; both stream(send) and AsyncFileASGIHandler send the transformed bytes. Keep middleware wrappers asynchronous to preserve bounded memory and backpressure. The original file or S3 body/client context closes on completion, read/send failure or cancellation, even when the wrapper does not forward iterator cleanup.

ASGI header names, including each separate set-cookie, are lowercase. Values retain Django's encoding and newline validation.

For custom adapters, override _stream_content() with an async generator that opens the resource in async with and yields chunks from _read_chunks(stream). The existing send_stream_to_client(stream, send) helper remains available for already-open streams; the caller owns that stream's context. Legacy adapters that open a resource in stream(send) and call this helper also send the middleware-wrapped iterator. Calling the helper binds the supplied stream lazily without bypassing middleware or reopening the built-in file/S3 source. Responses and their content iterators are single-use, as in Django.

class AsyncResponse[source]

AsyncResponse(*args, **kwargs) :: StreamingHttpResponse

Provides the is_async_fileresponse attribute checked by fileresponse.handlers.AsyncFileASGIHandler.

Contains the core content streaming logic which is the same for files and other file like streams.

Usage

class YourSpecialServiceResponse(AsyncResponse):
    async def _stream_content(self):
        async with your_service.open_stream() as stream:
            async for chunk in self._read_chunks(stream):
                yield chunk

Tests

Assert expected content is streamed by AsyncResponse

chunks = []


async def send(chunk):
    chunks.append(chunk)


class Stream:
    def __init__(self, data):
        self.index = 0
        self.data = data
        self.data_len = len(data)

    async def read(self, chunk_size):
        if self.index + chunk_size >= self.data_len:
            chunk = self.data[self.index :]
        else:
            chunk = self.data[self.index : self.index + chunk_size]
        self.index += chunk_size
        return chunk


content = "1234567890"
stream = Stream(content)
response = AsyncResponse(chunk_size=3)
await response.send_stream_to_client(stream, send)

sent_body = "".join([c["body"] for c in chunks if "body" in c])
assert content == sent_body

Make sure response_headers are set correctly

response = AsyncResponse(chunk_size=3)
assert len(response.response_headers) > 0

AiofileFileResponse

Serve a file asynchronously using aiofiles.

class AiofileFileResponse[source]

AiofileFileResponse(path, chunk_size=4096) :: AsyncResponse

Async response class for serving files from the filesystem. The async content iterator owns the open file context; AsyncResponse sends the iterator installed by middleware and closes the source on failure or cancellation.

It uses aiofiles to get the stream object from which chunks of data could be read asynchronously. The aiofiles library uses a threadpool to provide async filesystem access.

Usage

from fileresponse.http import AiofileFileResponse


async def aget_file(request, num=None):
    file_path = Path(__file__).parent.parent / "data" / str(num)
    return AiofileFileResponse(file_path, chunk_size=4096)

Tests

Complete FileResponse Testcase

• Create file with random content
• Stream file content via AsyncFileResponse
• Assert streamed content equals original content

import os

from pathlib import Path

path = Path("testdata.bin")

with path.open("wb") as f:
    f.write(os.urandom(10000))

with path.open("rb") as f:
    content = f.read()

chunk_datas = []


async def send(chunk_data):
    chunk_datas.append(chunk_data)


response = AiofileFileResponse(path)
await response.stream(send)
received_content = b"".join([cd.get("body", b"") for cd in chunk_datas])

assert content == received_content

sent size:  10000

test repr

response = AiofileFileResponse("foobar")
assert "<AiofileFileResponse status_code=200>" == repr(response)

AiobotocoreFileResponse

Serve a file asynchronously using aiobotocore.

The adapter uses the public aiobotocore.session.get_session API. Constructor overrides are resolved independently: endpoint_url, region_name, aws_access_key_id, secret_access_key and use_ssl. Each supplied option bypasses its Django S3 setting; omitted options fall back to the corresponding FILERESPONSE_S3_* setting. use_ssl defaults to False. Missing required settings raise ImproperlyConfigured at construction, before network access.

class AiobotocoreFileResponse[source]

AiobotocoreFileResponse(bucket, key, chunk_size=4096, **kwargs) :: AsyncResponse

Async response class for serving files from an s3 compatible object store like MinIO. Its async content iterator owns both the client and body contexts, including when middleware wraps the content.

It uses aiobotocore to get the stream object from which chunks of data could be read asynchronously.

Tests

import django


try:
    settings.configure()
except RuntimeError:
    pass


def configure_settings():
    settings.FILERESPONSE_S3_ENDPOINT_URL = "asdf"
    settings.FILERESPONSE_S3_REGION = "asdf"
    settings.FILERESPONSE_S3_ACCESS_KEY_ID = "asdf"
    settings.FILERESPONSE_S3_SECRET_ACCESS_KEY = "asdf"

Just test repr

configure_settings()
response = AiobotocoreFileResponse("bucket", "key")
assert "<AiobotocoreFileResponse status_code=200>" == repr(response)

Test endpoint url config

configure_settings()

expected_endpoint_url = "http://foobar:9000/"

settings.FILERESPONSE_S3_ENDPOINT_URL = "asdf"
response = AiobotocoreFileResponse("bucket", "key", endpoint_url=expected_endpoint_url)

assert expected_endpoint_url == response.client_config["endpoint_url"]