Mode now supports contextvars.ContextVar directly

faust/streams.py

@@ -4,6 +4,7 @@
 import typing
 import weakref
 from asyncio import CancelledError
+from contextvars import ContextVar
 from time import monotonic
 from typing import (
     Any,
@@ -26,6 +27,7 @@
 
 from mode import Seconds, Service, get_logger, want_seconds
 from mode.utils.aiter import aenumerate, aiter
+from mode.utils.compat import current_task
 from mode.utils.futures import maybe_async, notify
 from mode.utils.objects import cached_property
 from mode.utils.types.trees import NodeT
@@ -54,30 +56,6 @@
 
 logger = get_logger(__name__)
 
-try:  # pragma: no cover
-    from contextvars import ContextVar
-    from asyncio import current_task  # type: ignore
-
-    def _inherit_context(*, loop: asyncio.AbstractEventLoop = None) -> None:
-        ...
-except ImportError:  # pragma: no cover
-    from aiocontextvars import ContextVar, Context  # type: ignore
-
-    def _inherit_context(
-            *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
-        # see module: aiocontextvars.inherit
-        # this is the backport of the contextvars module added in CPython 3.7.
-        # it provides "thread-locals" for async generators, and asyncio.Task
-        # will automatically call this stuff in 3.7, but not in 3.6 so we call
-        # this when starting to iterate over the stream (Stream.__aiter__).
-        task = asyncio.Task.current_task(loop=loop or asyncio.get_event_loop())
-        # note: in actual CPython it's task._context, the aiocontextvars
-        # backport is a backport of a previous version of the PEP: :pep:`560`
-        task.ctx = Context(Context.current())  # type: ignore
-
-    current_task = asyncio.Task.current_task
-
-
 if typing.TYPE_CHECKING:  # pragma: no cover
     _current_event: ContextVar[weakref.ReferenceType[EventT]]
 _current_event = ContextVar('current_event')
@@ -691,7 +669,6 @@ def __next__(self) -> Any:
     async def __aiter__(self) -> AsyncIterator:
         self._finalized = True
         loop = self.loop
-        _inherit_context(loop=loop)
         await self.maybe_start()
         on_merge = self.on_merge
         on_stream_event_out = self._on_stream_event_out

requirements/README.rst

@@ -8,11 +8,6 @@
 
     These are the core dependencies required for Faust to work.
 
-+ ``py36.txt``
-
-    Additional dependencies to install when you do ``pip install faust``
-    and you are running Python 3.6.
-
 + ``extras/``
 
     Extra requirements for Faust features like RocksDB storage,

setup.py

@@ -105,13 +105,6 @@ def extras_require():
 else:
     long_description = 'See http://pypi.org/project/{}'.format(NAME)
 
-# -*- Install Requires -*-
-
-
-install_requires = reqs('default.txt')
-if sys.version_info < (3, 7):
-    install_requires += reqs('py36.txt')
-
 # -*- %%% -*-
 
 setup(
@@ -131,7 +124,7 @@ def extras_require():
     include_package_data=True,
     python_requires='>=3.6.0',
     zip_safe=False,
-    install_requires=install_requires,
+    install_requires=reqs('default.txt'),
     tests_require=reqs('test.txt'),
     extras_require=extras_require(),
     entry_points={

tox.ini

@@ -7,8 +7,6 @@ deps=
     -r{toxinidir}/requirements/test.txt
     -r{toxinidir}/requirements/ci.txt
 
-    3.6: -r{toxinidir}/requirements/py36.txt
-
     linkcheck,apicheck,configcheck: -r{toxinidir}/requirements/docs.txt
     flake8: -r{toxinidir}/requirements/dist.txt
 sitepackages = False