Repository: django/channels (python)

Issue:
asgiref dependency should be updated; channels 3.0.3 requires min. 3.2.10, which doesn't work for background workers
Channels 3.0.3 depends on `asgiref>=3.2.10`, however with that version, background workers will fail with `TypeError: __call__() missing 2 required positional arguments: 'receive' and 'send'` when receiving a message, even if declared with `.as_asgi()`.

Pull request:
Update asgiref dependency to >=3.3.1 (fixes #1613)
Update asgiref dependency to require at least version 3.3.1 which fixes crashing of background workers, as discussed in #1613.

Patch:
diff --git a/channels/auth.py b/channels/auth.py
index 3392552..93cd692 100644
--- a/channels/auth.py
+++ b/channels/auth.py
@@ -9,7 +9,6 @@ from django.contrib.auth import (
     user_logged_in,
     user_logged_out,
 )
-from django.contrib.auth.models import AnonymousUser
 from django.utils.crypto import constant_time_compare
 from django.utils.functional import LazyObject
 
@@ -24,6 +23,9 @@ def get_user(scope):
     Return the user model instance associated with the given scope.
     If no user is retrieved, return an instance of `AnonymousUser`.
     """
+    # postpone model import to avoid ImproperlyConfigured error before Django setup is complete.
+    from django.contrib.auth.models import AnonymousUser
+
     if "session" not in scope:
         raise ValueError(
             "Cannot find session in scope. You should wrap your consumer in "
@@ -112,6 +114,9 @@ def logout(scope):
     """
     Remove the authenticated user's ID from the request and flush their session data.
     """
+    # postpone model import to avoid ImproperlyConfigured error before Django setup is complete.
+    from django.contrib.auth.models import AnonymousUser
+
     if "session" not in scope:
         raise ValueError(
             "Login cannot find session in scope. You should wrap your "
diff --git a/channels/layers.py b/channels/layers.py
index fa9a9aa..5f26037 100644
--- a/channels/layers.py
+++ b/channels/layers.py
@@ -247,11 +247,11 @@ class InMemoryChannelLayer(BaseChannelLayer):
         queue = self.channels.setdefault(channel, asyncio.Queue())
 
         # Do a plain direct receive
-        _, message = await queue.get()
-
-        # Delete if empty
-        if queue.empty():
-            del self.channels[channel]
+        try:
+            _, message = await queue.get()
+        finally:
+            if queue.empty():
+                del self.channels[channel]
 
         return message
 
@@ -274,17 +274,14 @@ class InMemoryChannelLayer(BaseChannelLayer):
         """
         # Channel cleanup
         for channel, queue in list(self.channels.items()):
-            remove = False
             # See if it's expired
             while not queue.empty() and queue._queue[0][0] < time.time():
                 queue.get_nowait()
-                remove = True
-            # Any removal prompts group discard
-            if remove:
+                # Any removal prompts group discard
                 self._remove_from_groups(channel)
-            # Is the channel now empty and needs deleting?
-            if not queue:
-                del self.channels[channel]
+                # Is the channel now empty and needs deleting?
+                if queue.empty():
+                    del self.channels[channel]
 
         # Group Expiration
         timeout = int(time.time()) - self.group_expiry
diff --git a/channels/security/websocket.py b/channels/security/websocket.py
index 3f30833..eff745e 100644
--- a/channels/security/websocket.py
+++ b/channels/security/websocket.py
@@ -16,7 +16,7 @@ class OriginValidator:
         self.application = application
         self.allowed_origins = allowed_origins
 
-    async def __call__(self, scope, send, receive):
+    async def __call__(self, scope, receive, send):
         # Make sure the scope is of type websocket
         if scope["type"] != "websocket":
             raise ValueError(
@@ -34,11 +34,11 @@ class OriginValidator:
         # Check to see if the origin header is valid
         if self.valid_origin(parsed_origin):
             # Pass control to the application
-            return await self.application(scope, send, receive)
+            return await self.application(scope, receive, send)
         else:
             # Deny the connection
             denier = WebsocketDenier()
-            return await denier(scope, send, receive)
+            return await denier(scope, receive, send)
 
     def valid_origin(self, parsed_origin):
         """
@@ -95,10 +95,10 @@ class OriginValidator:
             return False
 
         # Get ResultParse object
-        parsed_pattern = urlparse(pattern.lower(), scheme=None)
+        parsed_pattern = urlparse(pattern.lower())
         if parsed_origin.hostname is None:
             return False
-        if parsed_pattern.scheme is None:
+        if not parsed_pattern.scheme:
             pattern_hostname = urlparse("//" + pattern).hostname or pattern
             return is_same_domain(parsed_origin.hostname, pattern_hostname)
         # Get origin.port or default ports for origin or None
diff --git a/docs/asgi.rst b/docs/asgi.rst
index d6b026e..0cea2fb 100644
--- a/docs/asgi.rst
+++ b/docs/asgi.rst
@@ -16,18 +16,15 @@ The full ASGI spec can be found at http://asgi.readthedocs.io
 Summary
 -------
 
-An ASGI application is a callable that takes a scope and returns a coroutine
-callable, that takes receive and send methods. It's usually written as a class:
+ASGI is structured as a single asynchronous callable, which takes a dict ``scope``
+and two callables ``receive`` and ``send``:
 
 .. code-block:: python
 
-    class Application:
-
-        def __init__(self, scope):
-            ...
-
-        async def __call__(self, receive, send):
-            ...
+    async def application(scope, receive, send):
+        event = await receive()
+        ...
+        await send({"type": "websocket.send", ...})
 
 The ``scope`` dict defines the properties of a connection, like its remote IP (for
 HTTP) or username (for a chat protocol), and the lifetime of a connection.
@@ -51,7 +48,7 @@ Composability
 
 ASGI applications, like WSGI ones, are designed to be composable, and this
 includes Channels' routing and middleware components like ``ProtocolTypeRouter``
-and ``SessionMiddeware``. These are just ASGI applications that take other
+and ``SessionMiddleware``. These are just ASGI applications that take other
 ASGI applications as arguments, so you can pass around just one top-level
 application for a whole Django project and dispatch down to the right consumer
 based on what sort of connection you're handling.
diff --git a/docs/channel_layer_spec.rst b/docs/channel_layer_spec.rst
index a75dff5..4c6b686 100644
--- a/docs/channel_layer_spec.rst
+++ b/docs/channel_layer_spec.rst
@@ -301,7 +301,7 @@ limitation that they only use the following characters:
 * Hyphen ``-``
 * Underscore ``_``
 * Period ``.``
-* Question mark ``?`` (only to delineiate single-reader channel names,
+* Question mark ``?`` (only to delineate single-reader channel names,
   and only one per name)
 * Exclamation mark ``!`` (only to delineate process-specific channel names,
   and only one per name)
diff --git a/docs/community.rst b/docs/community.rst
index 32e56ee..35f224d 100644
--- a/docs/community.rst
+++ b/docs/community.rst
@@ -11,6 +11,14 @@ These projects from the community are developed on top of Channels:
 * Apollo_, a real-time polling application for corporate and academic environments.
 * DjangoChannelsJsonRpc_, a wrapper for the JSON-RPC protocol.
 * channels-demultiplexer_, a (de)multiplexer for ``AsyncJsonWebsocketConsumer`` consumers.
+* channels_postgres_, a Django Channels channel layer that uses PostgreSQL as its backing store.
+
+Community Tutorials
+===================
+
+Here are some Channels tutorials from around the community:
+
+* kafka-integration_, a writeup on integrating Kafka with Channels.
 
 If you'd like to add your project, please submit a PR with a link and brief description.
 
@@ -22,3 +30,5 @@ If you'd like to add your project, please submit a PR with a link and brief desc
 .. _Apollo: https://github.com/maliesa96/apollo
 .. _DjangoChannelsJsonRpc: https://github.com/millerf/django-channels2-jsonrpc
 .. _channels-demultiplexer: https://github.com/csdenboer/channels-demultiplexer
+.. _kafka-integration: https://gist.github.com/aryan340/da071d027050cfe0a03df3b500f2f44b
+.. _channels_postgres: https://github.com/danidee10/channels_postgres
diff --git a/docs/deploying.rst b/docs/deploying.rst
index e3a63c9..f8060f7 100644
--- a/docs/deploying.rst
+++ b/docs/deploying.rst
@@ -18,36 +18,8 @@ is almost certainly going to be your top-level (Protocol Type) router.
 
 Here's an example of what that ``asgi.py`` might look like:
 
-.. code-block:: python
+.. include:: ./includes/asgi_example.rst
 
-    import os
-
-    from django.conf.urls import url
-    from django.core.asgi import get_asgi_application
-
-    # Fetch Django ASGI application early to ensure AppRegistry is populated
-    # before importing consumers and AuthMiddlewareStack that may import ORM
-    # models. 
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
-    django_asgi_app = get_asgi_application()
-    
-    from channels.auth import AuthMiddlewareStack
-    from channels.routing import ProtocolTypeRouter, URLRouter
-
-    from chat.consumers import AdminChatConsumer, PublicChatConsumer
-
-    application = ProtocolTypeRouter({
-        # Django's ASGI application to handle traditional HTTP requests
-        "http": django_asgi_app,
-
-        # WebSocket chat handler
-        "websocket": AuthMiddlewareStack(
-            URLRouter([
-                url(r"^chat/admin/$", AdminChatConsumer.as_asgi()),
-                url(r"^chat/$", PublicChatConsumer.as_asgi()),
-            ])
-        ),
-    })
 
 Setting up a channel backend
 ----------------------------
diff --git a/docs/includes/asgi_example.rst b/docs/includes/asgi_example.rst
new file mode 100644
index 0000000..b7c4b2e
--- /dev/null
+++ b/docs/includes/asgi_example.rst
@@ -0,0 +1,28 @@
+.. code-block:: python
+
+    import os
+
+    from channels.auth import AuthMiddlewareStack
+    from channels.routing import ProtocolTypeRouter, URLRouter
+    from django.conf.urls import url
+    from django.core.asgi import get_asgi_application
+
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
+    # Initialize Django ASGI application early to ensure the AppRegistry
+    # is populated before importing code that may import ORM models.
+    django_asgi_app = get_asgi_application()
+
+    from chat.consumers import AdminChatConsumer, PublicChatConsumer
+
+    application = ProtocolTypeRouter({
+        # Django's ASGI application to handle traditional HTTP requests
+        "http": django_asgi_app,
+
+        # WebSocket chat handler
+        "websocket": AuthMiddlewareStack(
+            URLRouter([
+                url(r"^chat/admin/$", AdminChatConsumer.as_asgi()),
+                url(r"^chat/$", PublicChatConsumer.as_asgi()),
+            ])
+        ),
+    })
\ No newline at end of file
diff --git a/docs/index.rst b/docs/index.rst
index 244f7fb..4e7a88a 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -53,6 +53,7 @@ Topics
    topics/testing
    topics/worker
    deploying
+   topics/troubleshooting
 
 
 Reference
diff --git a/docs/releases/2.3.0.rst b/docs/releases/2.3.0.rst
index 60c3d19..80023fa 100644
--- a/docs/releases/2.3.0.rst
+++ b/docs/releases/2.3.0.rst
@@ -14,7 +14,7 @@ Backwards Incompatible Changes
 ------------------------------
 
 As a result of the reworked body handling, ``AsgiRequest.__init__()`` is
-adjusted to expecte a file-like ``stream``, rather than the whole ``body`` as
+adjusted to expect a file-like ``stream``, rather than the whole ``body`` as
 bytes.
 
 Test cases instantiating requests directly will likely need to be updated to
diff --git a/docs/topics/routing.rst b/docs/topics/routing.rst
index f6df3b2..983e7ef 100644
--- a/docs/topics/routing.rst
+++ b/docs/topics/routing.rst
@@ -30,31 +30,8 @@ Channels projects and settings in :doc:`/deploying`.
 
 Here's an example of what that ``asgi.py`` might look like:
 
-.. code-block:: python
-
-    import os
-
-    from channels.auth import AuthMiddlewareStack
-    from channels.routing import ProtocolTypeRouter, URLRouter
-    from django.conf.urls import url
-    from django.core.asgi import get_asgi_application
-
-    from chat.consumers import AdminChatConsumer, PublicChatConsumer
+.. include:: ../includes/asgi_example.rst
 
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")
-
-    application = ProtocolTypeRouter({
-        # Django's ASGI application to handle traditional HTTP requests
-        "http": get_asgi_application()
-
-        # WebSocket chat handler
-        "websocket": AuthMiddlewareStack(
-            URLRouter([
-                url(r"^chat/admin/$", AdminChatConsumer.as_asgi()),
-                url(r"^chat/$", PublicChatConsumer.as_asgi()),
-            ])
-        ),
-    })
 
 .. note::
   We call the ``as_asgi()`` classmethod when routing our consumers. This
diff --git a/docs/topics/troubleshooting.rst b/docs/topics/troubleshooting.rst
new file mode 100644
index 0000000..2a70419
--- /dev/null
+++ b/docs/topics/troubleshooting.rst
@@ -0,0 +1,27 @@
+Troubleshooting
+===============
+
+
+
+ImproperlyConfigured exception
+------------------------------
+
+
+.. code-block:: text
+
+    django.core.exceptions.ImproperlyConfigured: Requested setting INSTALLED_APPS, but settings are not configured.
+    You must either define the environment variable DJANGO_SETTINGS_MODULE or call settings.configure() before accessing settings.
+
+
+This exception occurs when your application tries to import any models before Django finishes
+`its initialization process <https://docs.djangoproject.com/en/3.2/ref/applications/#initialization-process>`_ aka ``django.setup()``.
+
+
+``django.setup()`` `should be called only once <https://docs.djangoproject.com/en/3.2/topics/settings/#calling-django-setup-is-required-for-standalone-django-usage>`_,
+and should be called manually only in case of standalone apps.
+In context of Channels usage, ``django.setup()`` is called automatically in ``get_asgi_application()``,
+which means it needs to be called before any ORM models are imported.
+
+The working code order would look like this:
+
+.. include:: ../includes/asgi_example.rst
diff --git a/docs/tutorial/part_1.rst b/docs/tutorial/part_1.rst
index 4f8f9e1..a51d9f3 100644
--- a/docs/tutorial/part_1.rst
+++ b/docs/tutorial/part_1.rst
@@ -1,6 +1,10 @@
 Tutorial Part 1: Basic Setup
 ============================
 
+.. note::
+    If you encounter any issue during your coding session, please see the :doc:`/topics/troubleshooting` section.
+
+
 In this tutorial we will build a simple chat server. It will have two pages:
 
 * An index view that lets you type the name of a chat room to join.
diff --git a/setup.py b/setup.py
index c9bde0c..c8ba787 100644
--- a/setup.py
+++ b/setup.py
@@ -14,15 +14,14 @@ setup(
     python_requires='>=3.6',
     install_requires=[
         'Django>=2.2',
-        'asgiref>=3.2.10,<4',
+        'asgiref>=3.3.1,<4',
         'daphne>=3.0,<4',
     ],
     extras_require={
         'tests': [
             "pytest",
             "pytest-django",
-            "pytest-asyncio",
-            "async_generator",
+            "pytest-asyncio==0.14.0",
             "async-timeout",
             "coverage~=4.5",
         ],
diff --git a/tox.ini b/tox.ini
index ed1a519..aa8a2f2 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,6 +1,7 @@
 [tox]
 envlist =
-    py{36,37,38,39}-dj{22,30,31,master}
+    py{36,37,38,39}-dj{22,30,31,32}
+    py{38,39}-djmain
     qa
 
 [testenv]
@@ -12,7 +13,8 @@ deps =
     dj22: Django~=2.2.8
     dj30: Django==3.0.*
     dj31: Django>=3.1,<3.2
-    djmaster: https://github.com/django/django/archive/master.tar.gz
+    dj32: Django>=3.2b1,<4.0
+    djmain: https://github.com/django/django/archive/main.tar.gz
 
 [testenv:qa]
 skip_install=true


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index ef8f871..0b53e0a 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -21,17 +21,19 @@ jobs:
 
     steps:
       - uses: actions/checkout@v2
+
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v2
         with:
           python-version: ${{ matrix.python-version }}
+
       - name: Install dependencies
         run: |
-          python -m pip install --upgrade pip wheel setuptools tox
+          python -m pip install --upgrade pip wheel setuptools
+          python -m pip install --upgrade tox tox-py
+
       - name: Run tox targets for ${{ matrix.python-version }}
-        run: |
-          ENV_PREFIX=$(tr -C -d "0-9" <<< "${{ matrix.python-version }}")
-          TOXENV=$(tox --listenvs | grep "^py$ENV_PREFIX" | tr '\n' ',') python -m tox
+        run: tox --py current
 
   lint:
     name: Lint
diff --git a/tests/test_inmemorychannel.py b/tests/test_inmemorychannel.py
index 6544f7a..3f05ed7 100644
--- a/tests/test_inmemorychannel.py
+++ b/tests/test_inmemorychannel.py
@@ -2,20 +2,18 @@ import asyncio
 
 import async_timeout
 import pytest
-from async_generator import async_generator, yield_
 
 from channels.exceptions import ChannelFull
 from channels.layers import InMemoryChannelLayer
 
 
 @pytest.fixture()
-@async_generator
 async def channel_layer():
     """
     Channel layer fixture that flushes automatically.
     """
     channel_layer = InMemoryChannelLayer(capacity=3)
-    await yield_(channel_layer)
+    yield channel_layer
     await channel_layer.flush()
     await channel_layer.close()
 
@@ -106,3 +104,65 @@ async def test_groups_channel_full(channel_layer):
     await channel_layer.group_send("test-group", {"type": "message.1"})
     await channel_layer.group_send("test-group", {"type": "message.1"})
     await channel_layer.group_send("test-group", {"type": "message.1"})
+
+
+@pytest.mark.asyncio
+async def test_expiry_single():
+    """
+    Tests that a message can expire.
+    """
+    channel_layer = InMemoryChannelLayer(expiry=0.1)
+    await channel_layer.send("test-channel-1", {"type": "message.1"})
+    assert len(channel_layer.channels) == 1
+
+    await asyncio.sleep(0.1)
+
+    # Message should have expired and been dropped.
+    with pytest.raises(asyncio.TimeoutError):
+        async with async_timeout.timeout(0.5):
+            await channel_layer.receive("test-channel-1")
+
+    # Channel should be cleaned up.
+    assert len(channel_layer.channels) == 0
+
+
+@pytest.mark.asyncio
+async def test_expiry_unread():
+    """
+    Tests that a message on a channel can expire and be cleaned up even if
+    the channel is not read from again.
+    """
+    channel_layer = InMemoryChannelLayer(expiry=0.1)
+    await channel_layer.send("test-channel-1", {"type": "message.1"})
+
+    await asyncio.sleep(0.1)
+
+    await channel_layer.send("test-channel-2", {"type": "message.2"})
+    assert len(channel_layer.channels) == 2
+    assert (await channel_layer.receive("test-channel-2"))["type"] == "message.2"
+    # Both channels should be cleaned up.
+    assert len(channel_layer.channels) == 0
+
+
+@pytest.mark.asyncio
+async def test_expiry_multi():
+    """
+    Tests that multiple messages can expire.
+    """
+    channel_layer = InMemoryChannelLayer(expiry=0.1)
+    await channel_layer.send("test-channel-1", {"type": "message.1"})
+    await channel_layer.send("test-channel-1", {"type": "message.2"})
+    await channel_layer.send("test-channel-1", {"type": "message.3"})
+    assert (await channel_layer.receive("test-channel-1"))["type"] == "message.1"
+
+    await asyncio.sleep(0.1)
+    await channel_layer.send("test-channel-1", {"type": "message.4"})
+    assert (await channel_layer.receive("test-channel-1"))["type"] == "message.4"
+
+    # The second and third message should have expired and been dropped.
+    with pytest.raises(asyncio.TimeoutError):
+        async with async_timeout.timeout(0.5):
+            await channel_layer.receive("test-channel-1")
+
+    # Channel should be cleaned up.
+    assert len(channel_layer.channels) == 0