Support python 3.10 (#277)

.github/workflows/ci.yml

@@ -54,7 +54,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-         pyver: [3.6, 3.7, 3.8, 3.9]
+         pyver: ['3.6', '3.7', '3.8', '3.9', '3.10']
       fail-fast: true
     steps:
       - name: Checkhout

Makefile

@@ -1,7 +1,7 @@
 all: test
 
 isort:
-	isort -rc
+	isort .
 
 
 .install-deps: $(shell find requirements -type f)
@@ -15,7 +15,7 @@ isort:
 .flake: .install-deps .develop $(shell find aiojobs -type f) \
                       $(shell find tests -type f)
 	@flake8 .
-	@isort -rc -c
+	@isort -c .
 	@touch .flake
 
 flake: .flake

aiojobs/__init__.py

@@ -8,8 +8,6 @@
 
 __version__ = '0.3.0'
 
-import asyncio
-
 from ._scheduler import Scheduler
 
 
@@ -18,8 +16,7 @@ async def create_scheduler(*, close_timeout=0.1, limit=100,
     if exception_handler is not None and not callable(exception_handler):
         raise TypeError('A callable object or None is expected, '
                         'got {!r}'.format(exception_handler))
-    loop = asyncio.get_event_loop()
-    return Scheduler(loop=loop, close_timeout=close_timeout,
+    return Scheduler(close_timeout=close_timeout,
                      limit=limit, pending_limit=pending_limit,
                      exception_handler=exception_handler)
 

aiojobs/_job.py

@@ -11,8 +11,11 @@ class Job:
     _explicit = False
     _task = None
 
-    def __init__(self, coro, scheduler, loop):
-        self._loop = loop
+    def __init__(self, coro, scheduler):
+        if sys.version_info >= (3, 7):
+            self._loop = loop = asyncio.get_running_loop()
+        else:
+            self._loop = loop = asyncio.get_event_loop()
         self._coro = coro
         self._scheduler = scheduler
         self._started = loop.create_future()
@@ -44,7 +47,7 @@ def closed(self):
         return self._closed
 
     async def _do_wait(self, timeout):
-        async with async_timeout.timeout(timeout):
+        with async_timeout.timeout(timeout):
             # TODO: add a test for waiting for a pending coro
             await self._started
             return await self._task
@@ -55,8 +58,7 @@ async def wait(self, *, timeout=None):
         self._explicit = True
         scheduler = self._scheduler
         try:
-            return await asyncio.shield(self._do_wait(timeout),
-                                        loop=self._loop)
+            return await asyncio.shield(self._do_wait(timeout))
         except asyncio.CancelledError:
             # Don't stop inner coroutine on explicit cancel
             raise

aiojobs/_scheduler.py

@@ -1,28 +1,32 @@
 import asyncio
+import sys
 
 from ._job import Job
 
 try:
     from collections.abc import Collection
 except ImportError:  # pragma: no cover
     # Python 3.5 has no Collection ABC class
-    from collections.abc import Sized, Iterable, Container
+    from collections.abc import Container, Iterable, Sized
     bases = Sized, Iterable, Container
 else:  # pragma: no cover
     bases = (Collection,)
 
 
 class Scheduler(*bases):
     def __init__(self, *, close_timeout, limit, pending_limit,
-                 exception_handler, loop):
-        self._loop = loop
+                 exception_handler):
+        if sys.version_info >= (3, 7):
+            self._loop = loop = asyncio.get_running_loop()
+        else:
+            self._loop = loop = asyncio.get_event_loop()
         self._jobs = set()
         self._close_timeout = close_timeout
         self._limit = limit
         self._exception_handler = exception_handler
-        self._failed_tasks = asyncio.Queue(loop=loop)
+        self._failed_tasks = asyncio.Queue()
         self._failed_task = loop.create_task(self._wait_failed())
-        self._pending = asyncio.Queue(maxsize=pending_limit, loop=loop)
+        self._pending = asyncio.Queue(maxsize=pending_limit)
         self._closed = False
 
     def __iter__(self):
@@ -70,7 +74,7 @@ def closed(self):
     async def spawn(self, coro):
         if self._closed:
             raise RuntimeError("Scheduling a new job after closing")
-        job = Job(coro, self, self._loop)
+        job = Job(coro, self)
         should_start = (self._limit is None or
                         self.active_count < self._limit)
         self._jobs.add(job)
@@ -94,7 +98,7 @@ async def close(self):
                 self._pending.get_nowait()
             await asyncio.gather(
                 *[job._close(self._close_timeout) for job in jobs],
-                loop=self._loop, return_exceptions=True)
+                return_exceptions=True)
             self._jobs.clear()
         self._failed_tasks.put_nowait(None)
         await self._failed_task

pyproject.toml

@@ -22,12 +22,15 @@ classifiers = [
     "Programming Language :: Python :: 3.5",
     "Programming Language :: Python :: 3.6",
     "Programming Language :: Python :: 3.7",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
     "Topic :: Software Development",
 ]
 
 [tool.flit.metadata.requires-extra]
 aiohttp = [
-    "aiohttp >= 3.2.0"
+    "aiohttp >= 3.8.0"
 ]
 
 [tool.towncrier]

tests/conftest.py

@@ -13,7 +13,10 @@
 
 @pytest.fixture
 def scheduler(loop):
-    ret = Scheduler(loop=loop, **PARAMS)
+    async def maker():
+        return Scheduler(**PARAMS)
+
+    ret = loop.run_until_complete(maker())
     yield ret
     loop.run_until_complete(ret.close())
 

tests/test_aiohttp.py

@@ -1,14 +1,10 @@
 import asyncio
 
 import pytest
-
 from aiohttp import web
-from aiojobs.aiohttp import (
-    atomic,
-    get_scheduler,
-    get_scheduler_from_app,
-    get_scheduler_from_request,
-)
+
+from aiojobs.aiohttp import (atomic, get_scheduler, get_scheduler_from_app,
+                             get_scheduler_from_request)
 from aiojobs.aiohttp import setup as aiojobs_setup
 from aiojobs.aiohttp import spawn
 

tests/test_job.py

@@ -186,7 +186,7 @@ async def f():
 
     task = loop.create_task(job.wait())
     assert job.active, job
-    await asyncio.sleep(0.05, loop=loop)
+    await asyncio.sleep(0.05)
     assert job.active, job
     task.cancel()
     with suppress(asyncio.CancelledError):

tests/test_scheduler.py

@@ -1,17 +1,20 @@
 import asyncio
+import sys
 from unittest import mock
 
 import pytest
 from async_timeout import timeout
 
+from aiojobs import Scheduler
+
 
 def test_ctor(scheduler):
     assert len(scheduler) == 0
 
 
 async def test_spawn(scheduler, loop):
     async def coro():
-        await asyncio.sleep(1, loop=loop)
+        await asyncio.sleep(1)
     job = await scheduler.spawn(coro())
     assert not job.closed
 
@@ -39,7 +42,7 @@ async def test_exception_in_explicit_waiting(make_scheduler, loop):
     scheduler = await make_scheduler(exception_handler=exc_handler)
 
     async def coro():
-        await asyncio.sleep(0, loop=loop)
+        await asyncio.sleep(0)
         raise RuntimeError()
 
     job = await scheduler.spawn(coro())
@@ -61,13 +64,13 @@ async def test_exception_non_waited_job(make_scheduler, loop):
     exc = RuntimeError()
 
     async def coro():
-        await asyncio.sleep(0, loop=loop)
+        await asyncio.sleep(0)
         raise exc
 
     await scheduler.spawn(coro())
     assert len(scheduler) == 1
 
-    await asyncio.sleep(0.05, loop=loop)
+    await asyncio.sleep(0.05)
 
     assert len(scheduler) == 0
 
@@ -114,7 +117,7 @@ async def test_close_timeout(make_scheduler):
 
 async def test_scheduler_repr(scheduler, loop):
     async def coro():
-        await asyncio.sleep(1, loop=loop)
+        await asyncio.sleep(1)
 
     assert repr(scheduler) == '<Scheduler jobs=0>'
 
@@ -127,7 +130,7 @@ async def coro():
 
 async def test_close_jobs(scheduler, loop):
     async def coro():
-        await asyncio.sleep(1, loop=loop)
+        await asyncio.sleep(1)
 
     assert not scheduler.closed
 
@@ -165,7 +168,7 @@ def test_exception_handler_default(scheduler, loop):
 
 async def test_wait_with_timeout(scheduler, loop):
     async def coro():
-        await asyncio.sleep(1, loop=loop)
+        await asyncio.sleep(1)
 
     job = await scheduler.spawn(coro())
     with pytest.raises(asyncio.TimeoutError):
@@ -188,7 +191,7 @@ async def coro():
             await fut2
 
     job = await scheduler.spawn(coro())
-    await asyncio.sleep(0.001, loop=loop)
+    await asyncio.sleep(0.001)
     await scheduler.close()
     assert job.closed
     assert fut1.cancelled()
@@ -381,3 +384,15 @@ async def f():
 
     with pytest.warns(RuntimeWarning):
         del coro
+
+
+@pytest.mark.skipif(
+    sys.version_info < (3, 7),
+    reason="Python 3.6 doesn't support asyncio.get_running_loop()",
+)
+def test_scheduler_must_be_created_within_running_loop():
+    with pytest.raises(RuntimeError) as exc_info:
+        Scheduler(close_timeout=0, limit=0,
+                  pending_limit=0, exception_handler=None)
+
+    assert exc_info.match("no running event loop")