Django channels snippet

consumers.py

from channels.generic.websocket import JsonWebsocketConsumer
from django.contrib.auth import get_user_model

from users.serializers import UserResponseSerializer

User = get_user_model()


class UserConsumer(JsonWebsocketConsumer):
    def connect(self):
        self.accept()
        self.group_name = 'users_{}'.format(self.scope['url_route']['kwargs']['pk'])
        self.channel_layer.group_add(
            self.group_name, self.channel_name
        )

        self.users_post_save({
            'user_pk': self.scope['url_route']['kwargs']['pk']
        })

    def disconnect(self, close_code):
        self.channel_layer.group_discard(
            self.group_name, self.channel_name
        )

    def users_post_save(self, event):
        user = User.objects.get(
            pk=event.get('user_pk')
        )

        self.send_json(UserResponseSerializer(user).data)

models.py

from django.contrib.auth.models import AbstractUser
from channels.layers import get_channel_layer


class User(AbstractUser):
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)

        channel_layer = get_channel_layer()
        channel_layer.group_send(
            'users_{}'.format(self.pk), {
                'type': 'users.post_save',
                'user_pk': self.pk
            }
        )

requirements.txt

# I'm using Python 3.6.1

Django==1.11.7
channels==2.1.2
pytest==3.2.3
pytest-django==3.1.2

routing.py

from django.conf.urls import url

from . import consumers

websocket_urlpatterns = [
    url(
        r'^ws/users/(?P<pk>[0-9]+)/$',
        consumers.UserConsumer,
    ),
]

serializers.py

from django.contrib.auth import get_user_model

from rest_framework import serializers

User = get_user_model()


class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = '__all__'

test.py

import pytest

from decimal import Decimal

from channels.routing import URLRouter
from channels.testing import WebsocketCommunicator
from model_mommy import mommy

from users.routing import websocket_urlpatterns
from users.models import User


@pytest.fixture()
def user(db):
    return mommy.make(User)


@pytest.mark.asyncio
@pytest.mark.django_db
async def test_user_consumer(user):
    application = URLRouter(websocket_urlpatterns)
    communicator = WebsocketCommunicator(
        application,
        '/ws/users/{}/'.format(user.pk)
    )
    connected, subprotocol = await communicator.connect()
    assert connected
    assert subprotocol is None

    response = await communicator.receive_json_from()
    assert response == {
        'pk': user.pk,
        'username': user.username,
        'first_name': '',
        'last_name': '',
        'email': '',
    }
    await communicator.disconnect()

traceback

$ py.test -sv -k test_user_consumer

/Users/luan/Sources/env2channels-test/lib/python3.6/site-packages/daphne/server.py:12: UserWarning: Something has already installed a non-asyncio Twisted reactor. Attempting to uninstall it; you can fix this warning by importing daphne.server early in your codebase or finding the package that imports Twisted and importing it later on.
  UserWarning,
========================================================================================================== test session starts ==========================================================================================================
platform darwin -- Python 3.6.1, pytest-3.2.3, py-1.4.34, pluggy-0.4.0 -- /Users/luan/Sources/env2channels-test/bin/python3.6
cachedir: .cache
Django settings: caterpillar.settings (from ini file)
rootdir: /Users/luan/Sources/env2channels-test, inifile: pytest.ini
plugins: xdist-1.22.2, forked-0.2, django-3.1.2, cov-2.5.1, asyncio-0.8.0, celery-4.1.0
collected 277 items

users/tests/test_consumers.py::test_user_consumer Creating test database for alias 'default'...
PASSEDDestroying test database for alias 'default'...


=========================================================================================================== warnings summary ============================================================================================================
users/tests/test_consumers.py::test_user_consumer
  /Users/luan/Sources/channels-test/users/models.py:56: RuntimeWarning: coroutine 'RedisChannelLayer.group_send' was never awaited
    'user_pk': user.pk

-- Docs: http://doc.pytest.org/en/latest/warnings.html