Python 3.8 support

.ci/appveyor.yml

@@ -18,6 +18,8 @@ environment:
         - PYTHON: "C:\\Python36-x64\\python.exe"
         - PYTHON: "C:\\Python37\\python.exe"
         - PYTHON: "C:\\Python37-x64\\python.exe"
+        - PYTHON: "C:\\Python38\\python.exe"
+        - PYTHON: "C:\\Python38-x64\\python.exe"
 
 branches:
     # Avoid building PR branches.

.ci/travis-build-wheels.sh

@@ -39,9 +39,12 @@ pip install -U -r ".ci/requirements-publish.txt"
 if [ "${TRAVIS_OS_NAME}" == "linux" ]; then
     for pyver in ${RELEASE_PYTHON_VERSIONS}; do
         ML_PYTHON_VERSION=$(python3 -c \
-            "print('cp{maj}{min}-cp{maj}{min}m'.format( \
+            "print('cp{maj}{min}-cp{maj}{min}{s}'.format( \
                    maj='${pyver}'.split('.')[0], \
-                   min='${pyver}'.split('.')[1]))")
+                   min='${pyver}'.split('.')[1],
+                   s='m' if tuple('${pyver}'.split('.')) < ('3', '8') \
+                     else ''))")
+
 
         for arch in x86_64 i686; do
             ML_IMAGE="quay.io/pypa/manylinux1_${arch}"

.ci/travis-release.sh

@@ -30,8 +30,11 @@ P="${PYMODULE}-${PACKAGE_VERSION}"
 expected_wheels=()
 
 for pyver in ${RELEASE_PYTHON_VERSIONS}; do
-    pyver="${pyver//./}"
-    abitag="cp${pyver}-cp${pyver}m"
+    abitag=$(python -c \
+        "print('cp{maj}{min}-cp{maj}{min}{s}'.format( \
+                maj='${pyver}'.split('.')[0], \
+                min='${pyver}'.split('.')[1],
+                s='m' if tuple('${pyver}'.split('.')) < ('3', '8') else ''))")
     for plat in "${release_platforms[@]}"; do
         expected_wheels+=("${P}-${abitag}-${plat}.whl")
     done

.flake8

@@ -1,3 +1,3 @@
 [flake8]
-ignore = E402,E731
+ignore = E402,E731,W504,E252
 exclude = .git,__pycache__,build,dist,.eggs,.github,.local

.travis.yml

@@ -3,7 +3,7 @@ language: generic
 env:
     global:
         - PYMODULE=asyncpg
-        - RELEASE_PYTHON_VERSIONS="3.5 3.6 3.7"
+        - RELEASE_PYTHON_VERSIONS="3.5 3.6 3.7 3.8"
 
         - S3_UPLOAD_USERNAME=oss-ci-bot
         - S3_UPLOAD_BUCKET=magicstack-oss-releases
@@ -124,6 +124,15 @@ matrix:
           addons:
               apt: {packages: [postgresql-12]}
 
+        - os: linux
+          dist: xenial
+          sudo: true
+          language: python
+          python: "3.8"
+          env: BUILD=tests PGVERSION=12
+          addons:
+              apt: {packages: [postgresql-12]}
+
         # Build manylinux wheels.  Each wheel will be tested,
         # so there is no need for BUILD=tests here.
         # Also use this job to publish the releases and build
@@ -147,6 +156,9 @@ matrix:
         - os: osx
           env: BUILD=tests,wheels PYTHON_VERSION=3.7.4 PGVERSION=10
 
+        - os: osx
+          env: BUILD=tests,wheels PYTHON_VERSION=3.8.0 PGVERSION=10
+
 cache:
     pip
 

asyncpg/_testbase/__init__.py

@@ -80,7 +80,7 @@ def wrapper(self, *args, __meth__=meth, **kwargs):
                 coro = __meth__(self, *args, **kwargs)
                 timeout = getattr(__meth__, '__timeout__', mcls.TEST_TIMEOUT)
                 if timeout:
-                    coro = asyncio.wait_for(coro, timeout, loop=self.loop)
+                    coro = asyncio.wait_for(coro, timeout)
                     try:
                         self.loop.run_until_complete(coro)
                     except asyncio.TimeoutError:

asyncpg/_testbase/fuzzer.py

@@ -36,15 +36,13 @@ def __init__(self, *, listening_addr: str='127.0.0.1',
         self.listen_task = None
 
     async def _wait(self, work):
-        work_task = asyncio.ensure_future(work, loop=self.loop)
-        stop_event_task = asyncio.ensure_future(self.stop_event.wait(),
-                                                loop=self.loop)
+        work_task = asyncio.ensure_future(work)
+        stop_event_task = asyncio.ensure_future(self.stop_event.wait())
 
         try:
             await asyncio.wait(
                 [work_task, stop_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.stop_event.is_set():
                 raise StopServer()
@@ -58,7 +56,8 @@ async def _wait(self, work):
 
     def start(self):
         started = threading.Event()
-        self.thread = threading.Thread(target=self._start, args=(started,))
+        self.thread = threading.Thread(
+            target=self._start_thread, args=(started,))
         self.thread.start()
         if not started.wait(timeout=2):
             raise RuntimeError('fuzzer proxy failed to start')
@@ -70,13 +69,14 @@ def stop(self):
     def _stop(self):
         self.stop_event.set()
 
-    def _start(self, started_event):
+    def _start_thread(self, started_event):
         self.loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(self.loop)
 
-        self.connectivity = asyncio.Event(loop=self.loop)
+        self.connectivity = asyncio.Event()
         self.connectivity.set()
-        self.connectivity_loss = asyncio.Event(loop=self.loop)
-        self.stop_event = asyncio.Event(loop=self.loop)
+        self.connectivity_loss = asyncio.Event()
+        self.stop_event = asyncio.Event()
 
         if self.listening_port is None:
             self.listening_port = cluster.find_available_port()
@@ -92,15 +92,15 @@ def _start(self, started_event):
             self.loop.close()
 
     async def _main(self, started_event):
-        self.listen_task = asyncio.ensure_future(self.listen(), loop=self.loop)
+        self.listen_task = asyncio.ensure_future(self.listen())
         # Notify the main thread that we are ready to go.
         started_event.set()
         try:
             await self.listen_task
         finally:
             for c in list(self.connections):
                 c.close()
-            await asyncio.sleep(0.01, loop=self.loop)
+            await asyncio.sleep(0.01)
             if hasattr(self.loop, 'remove_reader'):
                 self.loop.remove_reader(self.sock.fileno())
             self.sock.close()
@@ -176,15 +176,15 @@ def close(self):
 
     async def handle(self):
         self.proxy_to_backend_task = asyncio.ensure_future(
-            self.proxy_to_backend(), loop=self.loop)
+            self.proxy_to_backend())
 
         self.proxy_from_backend_task = asyncio.ensure_future(
-            self.proxy_from_backend(), loop=self.loop)
+            self.proxy_from_backend())
 
         try:
             await asyncio.wait(
                 [self.proxy_to_backend_task, self.proxy_from_backend_task],
-                loop=self.loop, return_when=asyncio.FIRST_COMPLETED)
+                return_when=asyncio.FIRST_COMPLETED)
 
         finally:
             # Asyncio fails to properly remove the readers and writers
@@ -201,17 +201,14 @@ async def handle(self):
 
     async def _read(self, sock, n):
         read_task = asyncio.ensure_future(
-            self.loop.sock_recv(sock, n),
-            loop=self.loop)
+            self.loop.sock_recv(sock, n))
         conn_event_task = asyncio.ensure_future(
-            self.connectivity_loss.wait(),
-            loop=self.loop)
+            self.connectivity_loss.wait())
 
         try:
             await asyncio.wait(
                 [read_task, conn_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.connectivity_loss.is_set():
                 return None
@@ -225,15 +222,14 @@ async def _read(self, sock, n):
 
     async def _write(self, sock, data):
         write_task = asyncio.ensure_future(
-            self.loop.sock_sendall(sock, data), loop=self.loop)
+            self.loop.sock_sendall(sock, data))
         conn_event_task = asyncio.ensure_future(
-            self.connectivity_loss.wait(), loop=self.loop)
+            self.connectivity_loss.wait())
 
         try:
             await asyncio.wait(
                 [write_task, conn_event_task],
-                return_when=asyncio.FIRST_COMPLETED,
-                loop=self.loop)
+                return_when=asyncio.FIRST_COMPLETED)
 
             if self.connectivity_loss.is_set():
                 return None

asyncpg/connect_utils.py

@@ -531,7 +531,7 @@ async def _connect_addr(*, addr, loop, timeout, params, config,
     before = time.monotonic()
     try:
         tr, pr = await asyncio.wait_for(
-            connector, timeout=timeout, loop=loop)
+            connector, timeout=timeout)
     except asyncio.CancelledError:
         connector.add_done_callback(_close_leaked_connection)
         raise
@@ -540,8 +540,8 @@ async def _connect_addr(*, addr, loop, timeout, params, config,
     try:
         if timeout <= 0:
             raise asyncio.TimeoutError
-        await asyncio.wait_for(connected, loop=loop, timeout=timeout)
-    except Exception:
+        await asyncio.wait_for(connected, timeout=timeout)
+    except (Exception, asyncio.CancelledError):
         tr.close()
         raise
 
@@ -581,7 +581,7 @@ async def _negotiate_ssl_connection(host, port, conn_factory, *, loop, ssl,
     # accept SSLRequests. If the SSLRequest is accepted but either the SSL
     # negotiation fails or the PostgreSQL user isn't permitted to use SSL,
     # there's nothing that would attempt to reconnect with a non-SSL socket.
-    reader, writer = await asyncio.open_connection(host, port, loop=loop)
+    reader, writer = await asyncio.open_connection(host, port)
 
     tr = writer.transport
     try:
@@ -610,11 +610,13 @@ async def _negotiate_ssl_connection(host, port, conn_factory, *, loop, ssl,
 
         sock = sock.dup()  # Must come before tr.close()
     finally:
-        tr.close()
+        writer.close()
+        if hasattr(writer, 'wait_closed'):
+            await writer.wait_closed()
 
     try:
         return await conn_factory(sock=sock)  # Must come after tr.close()
-    except Exception:
+    except (Exception, asyncio.CancelledError):
         sock.close()
         raise
 
@@ -632,18 +634,18 @@ async def _create_ssl_connection(protocol_factory, host, port, *,
 
 async def _open_connection(*, loop, addr, params: _ConnectionParameters):
     if isinstance(addr, str):
-        r, w = await asyncio.open_unix_connection(addr, loop=loop)
+        r, w = await asyncio.open_unix_connection(addr)
     else:
         if params.ssl:
             r, w = await _negotiate_ssl_connection(
                 *addr,
-                functools.partial(asyncio.open_connection, loop=loop),
+                asyncio.open_connection,
                 loop=loop,
                 ssl=params.ssl,
                 server_hostname=addr[0],
                 ssl_is_advisory=params.ssl_is_advisory)
         else:
-            r, w = await asyncio.open_connection(*addr, loop=loop)
+            r, w = await asyncio.open_connection(*addr)
             _set_nodelay(_get_socket(w.transport))
 
     return r, w

asyncpg/connection.py

@@ -1080,7 +1080,7 @@ async def close(self, *, timeout=None):
         try:
             if not self.is_closed():
                 await self._protocol.close(timeout)
-        except Exception:
+        except (Exception, asyncio.CancelledError):
             # If we fail to close gracefully, abort the connection.
             self._abort()
             raise
@@ -1213,7 +1213,7 @@ async def _cancel(self, waiter):
             # the CancelledError, and don't want the loop to warn about
             # an unretrieved exception.
             pass
-        except Exception as ex:
+        except (Exception, asyncio.CancelledError) as ex:
             if not waiter.done():
                 waiter.set_exception(ex)
         finally:
@@ -1223,6 +1223,8 @@ async def _cancel(self, waiter):
                 waiter.set_result(None)
             if w is not None:
                 w.close()
+                if hasattr(w, 'wait_closed'):
+                    await w.wait_closed()
 
     def _cancel_current_command(self, waiter):
         self._cancellations.add(self._loop.create_task(self._cancel(waiter)))