Mixins

class CreateModelMixin[source]

Create model mixin.

async create(data, **kwargs)[source]

Create action.

Parameters:: data (dict) – model data to create.
Returns:: Tuple with the serializer data and the status code.
Return type:: Tuple[ReturnDict, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import CreateModelMixin

class LiveConsumer(CreateModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "create",
    request_id: new Date().getTime(),
    data: {
        username: "test",
        password1: "testpassword123",
        password2: "testpassword123",
    }
}))
/* The response will be something like this.
{
    "action": "create",
    "errors": [],
    "response_status": 201,
    "request_id": 150060530,
    "data": {'username': 'test', 'id': 42,},
}
*/

class DeleteModelMixin[source]

Delete model mixin

async delete(**kwargs)[source]

Retrieve action.

Returns:: Tuple with the serializer data and the status code.
Return type:: Tuple[None, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import DeleteModelMixin

class LiveConsumer(DeleteModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "delete",
    request_id: new Date().getTime(),
    pk: 1,
}))
/* The response will be something like this.
{
    "action": "delete",
    "errors": [],
    "response_status": 204,
    "request_id": 150000,
    "data": null,
}
*/

class ListModelMixin[source]

List model mixin

async list(**kwargs)[source]

List action.

Returns:: Tuple with the list of serializer data and the status code.
Return type:: Tuple[ReturnList, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import ListModelMixin

class LiveConsumer(ListModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "list",
    request_id: new Date().getTime(),
}))
/* The response will be something like this.
{
    "action": "list",
    "errors": [],
    "response_status": 200,
    "request_id": 1500000,
    "data": [
        {"email": "42@example.com", "id": 1, "username": "test1"},
        {"email": "45@example.com", "id": 2, "username": "test2"},
    ],
}
*/

class PaginatedModelListMixin[source]

A mixin that provides paginated listing functionality for model instances.

This mixin extends ListModelMixin to support paginated retrieval of model instances. It applies filtering, pagination, and serialization before returning the response.

permission_classes

The permission classes that control access to the list endpoint.

Type:: list

pagination_class

The pagination class used for paginating query results.

Type:: class

async list(**kwargs)[source]

Retrieves a paginated list of model instances.

This method applies queryset filtering, pagination, and serialization to return a paginated response. If pagination is disabled or not needed, it returns the full list of objects.

Parameters:: **kwargs – Additional keyword arguments used for filtering and serialization.

property paginator: any | None

Gets the paginator class

Returns:: Pagination class. Optional.

class PatchModelMixin[source]

Patch model mixin

async patch(data, **kwargs)[source]

Patch action.

Returns:: Tuple with the serializer data and the status code.
Parameters:: data (dict)
Return type:: Tuple[ReturnDict, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import PatchModelMixin

class LiveConsumer(PatchModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "patch",
    request_id: new Date().getTime(),
    pk: 1,
    data: {
        email: "00@example.com",
    },
}))
/* The response will be something like this.
{
    "action": "patch",
    "errors": [],
    "response_status": 200,
    "request_id": 150000,
    "data": {"email": "00@example.com", "id": 1, "username": "test1"},
}
*/

class RetrieveModelMixin[source]

Retrieve model mixin

async retrieve(**kwargs)[source]

Retrieve action.

Returns:: Tuple with the serializer data and the status code.
Return type:: Tuple[ReturnDict, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import RetrieveModelMixin

class LiveConsumer(RetrieveModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "retrieve",
    request_id: new Date().getTime(),
    pk: 1,
}))
/* The response will be something like this.
{
    "action": "retrieve",
    "errors": [],
    "response_status": 200,
    "request_id": 1500000,
    "data": {"email": "42@example.com", "id": 1, "username": "test1"},
}
*/

class StreamedPaginatedListMixin[source]

async list(action, request_id, **kwargs)[source]

Streams a paginated list of model instances.

This method retrieves a paginated list of objects, sends the initial response, and continues fetching and streaming additional pages until all results are delivered.

Parameters:

request_id (str) – A unique identifier for the request.
**kwargs – Additional keyword arguments used for pagination and filtering.
action (str)

class UpdateModelMixin[source]

Update model mixin

async update(data, **kwargs)[source]

Retrieve action.

Returns:: Tuple with the serializer data and the status code.
Parameters:: data (dict)
Return type:: Tuple[ReturnDict, int]

Examples

#! consumers.py
from .models import User
from .serializers import UserSerializer
from djangochannelsrestframework import permissions
from djangochannelsrestframework.generics import GenericAsyncAPIConsumer
from djangochannelsrestframework.mixins import UpdateModelMixin

class LiveConsumer(UpdateModelMixin, GenericAsyncAPIConsumer):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = (permissions.AllowAny,)

#! routing.py
from django.urls import re_path
from .consumers import LiveConsumer

websocket_urlpatterns = [
    re_path(r'^ws/$', LiveConsumer.as_asgi()),
]

// html
const ws = new WebSocket("ws://localhost:8000/ws/")
ws.send(JSON.stringify({
    action: "update",
    request_id: new Date().getTime(),
    pk: 1,
    data: {
        username: "test edited",
    },
}))
/* The response will be something like this.
{
    "action": "update",
    "errors": [],
    "response_status": 200,
    "request_id": 1500000,
    "data": {"email": "42@example.com", "id": 1, "username": "test edited"},
}
*/