Queue Mode in API¶

Queue mode is for long-running API work that should not block request/response. This guide shows how to run APIs asynchronously with Redis and Dramatiq. For architecture internals, see API Kit. Use this mode when endpoint runtime is high and clients should poll or stream progress instead of waiting for a direct response.

Prerequisites¶

redis-server installed locally or via docker
MOQueue model successfully migrated to DB under tbl_mo_queue.

1. Start Redis¶

Use one of the following:

# Local Redis (if installed)
redis-server

If you use Docker, make sure Docker is running.

# Docker
docker run --name mindoff-redis -p 6379:6379 redis:7

2. Ensure `REDIS_URL` Is Set¶

REDIS_URL is created during initialization and should be present in .env and loaded in config/settings.py:

REDIS_URL=redis://127.0.0.1:6379/0

REDIS_URL = config("REDIS_URL", default=None)

3. Start the Dramatiq Worker¶

From the same virtual environment as your project:

dramatiq django_mindoff.queue_worker

Implementation¶

1. Mindoff Queue API Class¶

Once Redis and the worker are running, queue mode is enabled by setting process_mode = "queue" on the API class.

Queue mode does not support file uploads

Multipart or file uploads are blocked when process_mode="queue".

Example queue-mode API configuration:

class CreateOrderV1APIView(MindoffAPIMixin):
    api_url_name = "orders__create_order"
    api_name = "Create Order"
    api_description = "Create a new order."
    method = "post"

    process_mode = "queue"
    allow_duplicate_queue = False
    progress_steps = {
        "validate": {"label": "Validated input", "percent": 10},
        "process": {"label": "Processed data", "percent": 60},
        "finalize": {"label": "Finalized response", "percent": 90},
    }

Queue-Specific API Attributes

These attributes only apply to queue-mode APIs. For the complete API attribute list, see API Development's Mindoff API Class.

Attribute	Purpose	Typical values
`process_mode`	Enable queue execution.	`queue`
`allow_duplicate_queue`	Allow multiple queued jobs per user with identical input.	`True` or `False`
`progress_steps`	Defines progress checkpoints and their percentages.	dict or `None`
`queue_detail_api_limit`	Rate limit for queue detail endpoint.	`30/m`
`queue_status_stream_api_limit`	Max concurrent status streams.	`3`
`queue_cancel_api_limit`	Rate limit for cancel endpoint.	`30/m`
`queue_retry_api_limit`	Rate limit for retry endpoint.	`30/m`

2. Progress Checkpoints¶

Record progress for a queue task at a checkpoint, updating the percent and label while validating the checkpoint key and checking for cancellation.

Usage:

class CreateOrderAPI(MindoffAPIMixin):
    process_mode = "queue"
    progress_steps = {
        "validate": {"label": "Validated input", "percent": 10},
        "process": {"label": "Processed data", "percent": 60},
        "finalize": {"label": "Finalized response", "percent": 90},
    }

    def run(self, request, *args, **kwargs):
        self.progress_checkpoint(request, "validate")
        self.progress_checkpoint(request, "process", msg="Rows processed")
        self.progress_checkpoint(request, "finalize")
        return mo_response_kit.json_response(
            code="SUCCESS",
            category="success",
        )

Parameters:

request (Any): Current request object. Queue context is read from request.queue_task_uuid.
checkpoint_key (str): Key in class-level progress_steps mapped to a step config containing label and percent.
msg (str | None, default=None): Optional progress message. When omitted, configured step label is used.

Possible responses:

Returns None when request is not running in queue context.
Updates queue progress when checkpoint key is valid.
Raises MindoffValidationError when queue cancellation is requested.
Raises validation error when checkpoint key is not configured.

Notes:

Intended for process_mode="queue" APIs.
progress_steps should follow: {"step_key": {"label": "...", "percent": int}}.
Use it at meaningful checkpoints so clients can show progress between queued and completed.

Core Concepts¶

1. Queue Response Format¶

When an API runs in queue mode, the initial response is a queued acknowledgment:

{
    "status": "ok",
    "message": {
        "code": "QUEUED",
        "title": "Queued",
        "description": "Operation queued successfully.",
        "category": "success"
    },
    "data": {
        "queue_id": "<uuid>",
        "response_url": "<absolute-url>",
        "status_stream_url": "<absolute-url>",
        "stream_ticket_url": "<absolute-url>",
        "cancel_url": "<absolute-url>",
        "retry_url": "<absolute-url>",
        "progress_steps": {}
    }
}

Use response_url to fetch the final result, or status_stream_url for live progress updates. stream_ticket_url authenticates that stream from a browser (see Authentication and Ownership). progress_steps is echoed back so clients can display expected checkpoints.

2. Built-in APIs to Manage Queue¶

Once a task is queued, use these endpoints to retrieve results, monitor status, or control execution:

Endpoint	Method	Purpose	Notes
`queue/<uuid:queue_task_uuid>/`	`GET`	Retrieve the result (or current status) of a queue task.	Returns final response when completed.
`queue/list/`	`GET`	List queue tasks with optional filtering.	Query params below.
`queue/<uuid:queue_task_uuid>/stream/`	`GET`	Real-time queue task status via SSE.	`text/event-stream`.
`queue/<uuid:queue_task_uuid>/stream-ticket/`	`POST`	Issue a ticket that authenticates an SSE stream.	Single-use, expires in seconds.
`queue/<uuid:queue_task_uuid>/cancel/`	`POST`	Cancel a running or queued task.	Returns cancel requested or not cancellable.
`queue/<uuid:queue_task_uuid>/retry/`	`POST`	Retry a previously failed queue task.	Returns new queued response.

Queue list query params (all optional, combinable):

id
job_status
user_ref_id
owner_id
api_url_name
page
page_size

Results are scoped to the caller, and these params filter within that scope — they can never widen it. See Authentication and Ownership.

3. Authentication and Ownership¶

These endpoints are generic: they serve tasks created by any queue-mode API, each of which may use a different authentication scheme. So a request to a task endpoint is authenticated with the authentication_classes of the API that created that task, resolved from the task's api_url_name. Nothing extra to configure — an API behind JWTAuthentication gets queue endpoints behind JWTAuthentication.

A task is then readable only by the user who queued it:

Task has an owner — only that user may read, stream, cancel, or retry it. Everyone else receives PERMISSION_DENIED.
Task has no owner — queued anonymously, because the originating API allows anonymous access. It stays anonymously reachable by queue_task_uuid.
Originating API no longer exists — renamed or removed since the task was queued. Authentication falls back to DRF's DEFAULT_AUTHENTICATION_CLASSES, so an authenticated owner still gets through and an anonymous caller does not.

queue/list/ carries no task id and so has no originating API to borrow from. It uses MINDOFF_QUEUE_LIST_AUTHENTICATION_CLASSES when set and DRF's DEFAULT_AUTHENTICATION_CLASSES otherwise, and scopes its results: staff see every task, an authenticated user sees their own, and an anonymous caller sees only the ownerless tasks queued by their own session.

Streaming from a Browser¶

A browser EventSource cannot set an Authorization header. If your APIs authenticate by cookie (SessionAuthentication), the stream needs nothing special — the cookie is sent automatically.

For token-based clients, exchange the token for a stream ticket instead of putting it in the URL. The ticket is single-use, expires in seconds, and is bound to one task and one user, so it does not matter that URLs end up in access logs and browser history — which is exactly why a long-lived token must never be placed there (RFC 6750 §2.3).

const res = await fetch(`/mindoff/queue/${queueId}/stream-ticket/`, {
    method: "POST",
    headers: { Authorization: `Bearer ${accessToken}` },
});
const { data } = await res.json();

// data.status_stream_url already carries the ticket
const events = new EventSource(data.status_stream_url);
events.onmessage = (e) => console.log(JSON.parse(e.data));

The ticket lifetime defaults to 30 seconds and is set with MINDOFF_QUEUE_STREAM_TICKET_TTL. Request a ticket immediately before opening the stream, and request a fresh one to reconnect. If the originating API uses SessionAuthentication, this POST needs a CSRF token like any other session authenticated POST.

4. MOQueue Model and Task Persistence¶

Queue tasks are persisted in the MOQueue model. This provides durable storage for:

job_status
request snapshot
response payload and HTTP status
error payload (if failed)

Redis stores live progress for running tasks and feeds the SSE stream. The queue detail endpoint reads from the database, while the status stream reads live state from Redis.

Troubleshooting¶

Redis-related errors in queue mode
Set REDIS_URL correctly and ensure Redis is running locally or reachable from your host.

MOQueue table missing or errors about tbl_mo_queue
Run:

python manage.py makemigrations django_mindoff
python manage.py migrate