Mixins

class CreateModelMixin

Create model mixin.

async create(data, **kwargs)

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

Delete model mixin

async delete(**kwargs)

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

List model mixin

async list(**kwargs)

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

async list(**kwargs)

List action.

Returns

Tuple with the list of serializer data and the status code.

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"},
    ],
}
*/

property paginator: Optional[any]

Gets the paginator class

Returns

Pagination class. Optional.

class PatchModelMixin

Patch model mixin

async patch(data, **kwargs)

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

Retrieve model mixin

async retrieve(**kwargs)

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

async list(action, request_id, **kwargs)

List action.

Returns

Tuple with the list of serializer data and the status code.

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 UpdateModelMixin

Update model mixin

async update(data, **kwargs)

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"},
}
*/