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
 

Makefile

@@ -34,7 +34,6 @@ test:
 
 testinstalled:
 	cd /tmp && $(PYTHON) $(ROOT)/tests/__init__.py
-	cd /tmp && USE_UVLOOP=1 $(PYTHON) $(ROOT)/tests/__init__.py
 
 
 quicktest:

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/compat.py

@@ -79,3 +79,14 @@ def current_asyncio_task(loop):
 else:
     def current_asyncio_task(loop):
         return asyncio.Task.current_task(loop)
+
+
+async def wait_closed(stream):
+    # Not all asyncio versions have StreamWriter.wait_closed().
+    if hasattr(stream, 'wait_closed'):
+        try:
+            await stream.wait_closed()
+        except ConnectionResetError:
+            # On Windows wait_closed() sometimes propagates
+            # ConnectionResetError which is totally unnecessary.
+            pass