Repository: django/asgiref
Pull request #233 (2021-01-24 19:43:31)

 type check annotations  (and drop support for python 3.5)


Patch:
diff --git a/README.rst b/README.rst
index 259f31a..f99aee6 100644
--- a/README.rst
+++ b/README.rst
@@ -96,7 +96,7 @@ file handles for incoming POST bodies).
 Dependencies
 ------------
 
-``asgiref`` requires Python 3.5 or higher.
+``asgiref`` requires Python 3.6 or higher.
 
 
 Contributing
@@ -163,14 +163,9 @@ The first compromise you get to might be that ``thread_sensitive`` code should
 just run in the same thread and not spawn in a sub-thread, fulfilling the first
 restriction, but that immediately runs you into the second restriction.
 
-By default, a variant of ThreadPoolExecutor executes any ``thread_sensitive``
-code on the outermost synchronous thread - either the main thread, or a single
-spawned subthread.
-
-There is an option to override the default single-threaded behavior so that
-there is 1 synchronous thread per context. If ``current_context_func`` is
-specified, this function will be called to retrieve the current context. There
-will be exactly 1 synchronous thread per context in this case.
+The only real solution is to essentially have a variant of ThreadPoolExecutor
+that executes any ``thread_sensitive`` code on the outermost synchronous
+thread - either the main thread, or a single spawned subthread.
 
 This means you now have two basic states:
 
diff --git a/asgiref/local.py b/asgiref/local.py
index 3d24b14..bb69f73 100644
--- a/asgiref/local.py
+++ b/asgiref/local.py
@@ -32,10 +32,10 @@ class Local:
 
     CLEANUP_INTERVAL = 60  # seconds
 
-    def __init__(self, thread_critical=False):
+    def __init__(self, thread_critical: bool = False) -> None:
         self._thread_critical = thread_critical
         self._thread_lock = threading.RLock()
-        self._context_refs = weakref.WeakSet()
+        self._context_refs: "weakref.WeakSet[object]" = weakref.WeakSet()
         # Random suffixes stop accidental reuse between different Locals,
         # though we try to force deletion as well.
         self._attr_name = "_asgiref_local_impl_%s_%s" % (
diff --git a/asgiref/sync.py b/asgiref/sync.py
index d62ee86..f07f27e 100644
--- a/asgiref/sync.py
+++ b/asgiref/sync.py
@@ -1,4 +1,3 @@
-import asyncio
 import asyncio.coroutines
 import functools
 import os
@@ -6,13 +5,14 @@ import sys
 import threading
 import weakref
 from concurrent.futures import Future, ThreadPoolExecutor
+from typing import Dict
 
 from .current_thread_executor import CurrentThreadExecutor
 from .local import Local
 
-try:
-    import contextvars  # Python 3.7+ only.
-except ImportError:
+if sys.version_info >= (3, 7):
+    import contextvars
+else:
     contextvars = None
 
 
@@ -27,6 +27,59 @@ def _restore_context(context):
             cvar.set(context.get(cvar))
 
 
+class ThreadSensitiveContext:
+    """Async context manager to manage context for thread sensitive mode
+
+    This context manager controls which thread pool executor is used when in
+    thread sensitive mode. By default, a single thread pool executor is shared
+    within a process.
+
+    In Python 3.7+, the ThreadSensitiveContext() context manager may be used to
+    specify a thread pool per context.
+
+    In Python 3.6, usage of this context manager has no effect.
+
+    This context manager is re-entrant, so only the outer-most call to
+    ThreadSensitiveContext will set the context.
+
+    Usage:
+
+    >>> import time
+    >>> async with ThreadSensitiveContext():
+    ...     await sync_to_async(time.sleep, 1)()
+    """
+
+    def __init__(self):
+        self.token = None
+
+    if contextvars:
+
+        async def __aenter__(self):
+            try:
+                SyncToAsync.thread_sensitive_context.get()
+            except LookupError:
+                self.token = SyncToAsync.thread_sensitive_context.set(self)
+
+            return self
+
+        async def __aexit__(self, exc, value, tb):
+            if not self.token:
+                return
+
+            executor = SyncToAsync.context_to_thread_executor.pop(self, None)
+            if executor:
+                executor.shutdown()
+            SyncToAsync.thread_sensitive_context.reset(self.token)
+
+    else:
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc, value, tb):
+            pass
+
+
 class AsyncToSync:
     """
     Utility class which turns an awaitable that only works on the thread with
@@ -41,7 +94,7 @@ class AsyncToSync:
     """
 
     # Maps launched Tasks to the threads that launched them (for locals impl)
-    launch_map = {}
+    launch_map: "Dict[asyncio.Task[object], threading.Thread]" = {}
 
     # Keeps track of which CurrentThreadExecutor to use. This uses an asgiref
     # Local, not a threadlocal, so that tasks can work out what their parent used.
@@ -232,12 +285,6 @@ class SyncToAsync:
     outer code. This is needed for underlying Python code that is not
     threadsafe (for example, code which handles SQLite database connections).
 
-    If current_context_func is passed, the code will run 1 thread per context.
-    As an example, this may be used to create a per-request synchronous thread
-    by specifying the request object as the context. Thread scheduling will
-    occur by request in this scenario - each request will execute synchronous
-    work within the same thread.
-
     If the outermost program is async (i.e. SyncToAsync is outermost), then
     this will be a dedicated single sub-thread that all sync code runs in,
     one after the other. If the outermost program is sync (i.e. AsyncToSync is
@@ -254,7 +301,7 @@ class SyncToAsync:
         )
 
     # Maps launched threads to the coroutines that spawned them
-    launch_map = {}
+    launch_map: "Dict[threading.Thread, asyncio.Task[object]]" = {}
 
     # Storage for main event loop references
     threadlocal = threading.local()
@@ -262,16 +309,23 @@ class SyncToAsync:
     # Single-thread executor for thread-sensitive code
     single_thread_executor = ThreadPoolExecutor(max_workers=1)
 
+    # Maintain a contextvar for the current execution context. Optionally used
+    # for thread sensitive mode.
+    thread_sensitive_context = (
+        contextvars.ContextVar("thread_sensitive_context") if contextvars else None
+    )
+
     # Maintaining a weak reference to the context ensures that thread pools are
     # erased once the context goes out of scope. This terminates the thread pool.
-    context_to_thread_executor = weakref.WeakKeyDictionary()
+    context_to_thread_executor: "weakref.WeakKeyDictionary[object, ThreadPoolExecutor]" = (
+        weakref.WeakKeyDictionary()
+    )
 
-    def __init__(self, func, thread_sensitive=True, current_context_func=None):
+    def __init__(self, func, thread_sensitive=True):
         self.func = func
         functools.update_wrapper(self, func)
         self._thread_sensitive = thread_sensitive
         self._is_coroutine = asyncio.coroutines._is_coroutine
-        self._current_context_func = current_context_func
         try:
             self.__self__ = func.__self__
         except AttributeError:
@@ -285,18 +339,20 @@ class SyncToAsync:
             if hasattr(AsyncToSync.executors, "current"):
                 # If we have a parent sync thread above somewhere, use that
                 executor = AsyncToSync.executors.current
-            elif self._current_context_func:
+            elif self.thread_sensitive_context and self.thread_sensitive_context.get(
+                None
+            ):
                 # If we have a way of retrieving the current context, attempt
                 # to use a per-context thread pool executor
-                current_context = self._current_context_func()
+                thread_sensitive_context = self.thread_sensitive_context.get()
 
-                if current_context in self.context_to_thread_executor:
+                if thread_sensitive_context in self.context_to_thread_executor:
                     # Re-use thread executor in current context
-                    executor = self.context_to_thread_executor[current_context]
+                    executor = self.context_to_thread_executor[thread_sensitive_context]
                 else:
                     # Create new thread executor in current context
                     executor = ThreadPoolExecutor(max_workers=1)
-                    self.context_to_thread_executor[current_context] = executor
+                    self.context_to_thread_executor[thread_sensitive_context] = executor
             else:
                 # Otherwise, we run it in a fixed single thread
                 executor = self.single_thread_executor
@@ -393,15 +449,13 @@ class SyncToAsync:
 async_to_sync = AsyncToSync
 
 
-def sync_to_async(func=None, thread_sensitive=True, current_context_func=None):
+def sync_to_async(func=None, thread_sensitive=True):
     if func is None:
         return lambda f: SyncToAsync(
             f,
             thread_sensitive=thread_sensitive,
-            current_context_func=current_context_func,
         )
     return SyncToAsync(
         func,
         thread_sensitive=thread_sensitive,
-        current_context_func=current_context_func,
     )
diff --git a/asgiref/timeout.py b/asgiref/timeout.py
index 0ff5892..201d017 100644
--- a/asgiref/timeout.py
+++ b/asgiref/timeout.py
@@ -9,9 +9,7 @@
 import asyncio
 import sys
 from types import TracebackType
-from typing import Any, Optional, Type  # noqa
-
-PY_37 = sys.version_info >= (3, 7)
+from typing import Any, Optional, Type
 
 
 class timeout:
@@ -115,14 +113,15 @@ class timeout:
             self._cancelled = True
 
 
-def current_task(loop: asyncio.AbstractEventLoop) -> "asyncio.Task[Any]":
-    if PY_37:
-        task = asyncio.current_task(loop=loop)  # type: ignore
+def current_task(loop: asyncio.AbstractEventLoop) -> "Optional[asyncio.Task[Any]]":
+    if sys.version_info >= (3, 7):
+        task = asyncio.current_task(loop=loop)
     else:
         task = asyncio.Task.current_task(loop=loop)
     if task is None:
         # this should be removed, tokio must use register_task and family API
-        if hasattr(loop, "current_task"):
-            task = loop.current_task()  # type: ignore
+        fn = getattr(loop, "current_task", None)
+        if fn is not None:
+            task = fn()
 
     return task
diff --git a/docs/conf.py b/docs/conf.py
index 6935f11..34403a4 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from typing import List, Dict
 #
 # ASGI documentation build configuration file, created by
 # sphinx-quickstart on Thu May 17 21:22:10 2018.
@@ -31,10 +32,10 @@
 # Add any Sphinx extension module names here, as strings. They can be
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
-extensions = []
+extensions: List[str] = []
 
 # Add any paths that contain templates here, relative to this directory.
-templates_path = []
+templates_path: List[str] = []
 
 # The suffix(es) of source filenames.
 # You can specify multiple suffix as a list of string:
@@ -120,7 +121,7 @@ htmlhelp_basename = 'ASGIdoc'
 
 # -- Options for LaTeX output ---------------------------------------------
 
-latex_elements = {
+latex_elements: Dict[str, str] = {
     # The paper size ('letterpaper' or 'a4paper').
     #
     # 'papersize': 'letterpaper',
diff --git a/setup.cfg b/setup.cfg
index 0bd7dc9..a3df063 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -16,7 +16,6 @@ classifiers =
     Programming Language :: Python
     Programming Language :: Python :: 3
     Programming Language :: Python :: 3 :: Only
-    Programming Language :: Python :: 3.5
     Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
@@ -28,7 +27,7 @@ project_urls =
     Changelog = https://github.com/django/asgiref/blob/master/CHANGELOG.txt
 
 [options]
-python_requires = >=3.5
+python_requires = >=3.6
 packages = find:
 include_package_data = true
 install_requires =
@@ -39,6 +38,7 @@ zip_safe = false
 tests =
     pytest
     pytest-asyncio
+    mypy>=0.800
 
 [tool:pytest]
 testpaths = tests
@@ -50,3 +50,63 @@ max-line-length = 119
 
 [isort]
 line_length = 119
+
+[mypy]
+warn_unused_ignores = True
+strict = True
+
+[mypy-asgiref.current_thread_executor]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.local]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.sync]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.compatibility]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.wsgi]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.testing]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-asgiref.server]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_server]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_wsgi]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_testing]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_sync_contextvars]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_sync]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_local]
+disallow_untyped_defs = False
+check_untyped_defs = False
+
+[mypy-test_compatibility]
+disallow_untyped_defs = False
+check_untyped_defs = False
diff --git a/setup.py b/setup.py
index 3363a97..f2dbecd 100644
--- a/setup.py
+++ b/setup.py
@@ -1,3 +1,3 @@
-from setuptools import setup
+from setuptools import setup  # type: ignore[import]
 
 setup(name='asgiref')
diff --git a/tox.ini b/tox.ini
index 2523c24..5b518fe 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,13 +1,14 @@
 [tox]
 envlist =
-    py{35,36,37,38,39}
+    py{36,37,38,39}-{test,mypy}
     qa
 
 [testenv]
 usedevelop = true
 extras = tests
 commands =
-    pytest -v {posargs}
+    test: pytest -v {posargs}
+    mypy: mypy . {posargs}
 
 [testenv:qa]
 skip_install=true


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 6737a86..3bc8d80 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -14,7 +14,6 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-        - 3.5
         - 3.6
         - 3.7
         - 3.8
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 168cb39..5e9c171 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -8,7 +8,7 @@ from unittest import TestCase
 
 import pytest
 
-from asgiref.sync import async_to_sync, sync_to_async
+from asgiref.sync import ThreadSensitiveContext, async_to_sync, sync_to_async
 
 
 @pytest.mark.asyncio
@@ -323,54 +323,54 @@ async def test_thread_sensitive_outside_async():
 
 
 @pytest.mark.asyncio
-async def test_thread_sensitive_with_context_different():
+async def test_thread_sensitive_with_context_matches():
     result_1 = {}
     result_2 = {}
 
-    class Context:
-        pass
-
-    context_1 = Context()
-    context_2 = Context()
-
     def store_thread(result):
         result["thread"] = threading.current_thread()
 
-    fn_1 = sync_to_async(store_thread, current_context_func=lambda: context_1)
-    fn_2 = sync_to_async(store_thread, current_context_func=lambda: context_2)
+    store_thread_async = sync_to_async(store_thread)
+
+    async def fn():
+        async with ThreadSensitiveContext():
+            # Run it (in supposed parallel!)
+            await asyncio.wait(
+                [store_thread_async(result_1), store_thread_async(result_2)]
+            )
 
-    # Run it (in true parallel!)
-    await asyncio.wait([fn_1(result_1), fn_2(result_2)])
+    await fn()
 
-    # They should not have run in the main thread, and on different threads
+    # They should not have run in the main thread, and on the same threads
     assert result_1["thread"] != threading.current_thread()
-    assert result_1["thread"] != result_2["thread"]
+    assert result_1["thread"] == result_2["thread"]
 
 
 @pytest.mark.asyncio
-async def test_thread_sensitive_with_context_matches():
+async def test_thread_sensitive_nested_context():
     result_1 = {}
     result_2 = {}
 
-    class Context:
-        pass
-
-    context = Context()
-
+    @sync_to_async
     def store_thread(result):
         result["thread"] = threading.current_thread()
 
-    fn_1 = sync_to_async(store_thread, current_context_func=lambda: context)
-    fn_2 = sync_to_async(store_thread, current_context_func=lambda: context)
-
-    # Run it (in supposed parallel!)
-    await asyncio.wait([fn_1(result_1), fn_2(result_2)])
+    async with ThreadSensitiveContext():
+        await store_thread(result_1)
+        async with ThreadSensitiveContext():
+            await store_thread(result_2)
 
-    # They should not have run in the main thread, and on different threads
+    # They should not have run in the main thread, and on the same threads
     assert result_1["thread"] != threading.current_thread()
     assert result_1["thread"] == result_2["thread"]
 
 
+@pytest.mark.asyncio
+async def test_thread_sensitive_context_without_sync_work():
+    async with ThreadSensitiveContext():
+        pass
+
+
 def test_thread_sensitive_double_nested_sync():
     """
     Tests that thread_sensitive SyncToAsync nests inside itself where the
diff --git a/tests/test_sync_contextvars.py b/tests/test_sync_contextvars.py
index af3e3be..b1027aa 100644
--- a/tests/test_sync_contextvars.py
+++ b/tests/test_sync_contextvars.py
@@ -1,15 +1,37 @@
 import asyncio
+import threading
 import time
 
 import pytest
 
-from asgiref.sync import async_to_sync, sync_to_async
+from asgiref.sync import ThreadSensitiveContext, async_to_sync, sync_to_async
 
 contextvars = pytest.importorskip("contextvars")
 
 foo = contextvars.ContextVar("foo")
 
 
+@pytest.mark.asyncio
+async def test_thread_sensitive_with_context_different():
+    result_1 = {}
+    result_2 = {}
+
+    @sync_to_async
+    def store_thread(result):
+        result["thread"] = threading.current_thread()
+
+    async def fn(result):
+        async with ThreadSensitiveContext():
+            await store_thread(result)
+
+    # Run it (in true parallel!)
+    await asyncio.wait([fn(result_1), fn(result_2)])
+
+    # They should not have run in the main thread, and on different threads
+    assert result_1["thread"] != threading.current_thread()
+    assert result_1["thread"] != result_2["thread"]
+
+
 @pytest.mark.asyncio
 async def test_sync_to_async_contextvars():
     """