Add Windows support (#31)

Author: elprans

.ci/appveyor.yml

@@ -0,0 +1,35 @@
+services:
+    - postgresql95
+
+environment:
+    global:
+        PGINSTALLATION: C:\\Program Files\\PostgreSQL\\9.5\\bin
+        TWINE_USERNAME: magicstack-ci
+        TWINE_PASSWORD:
+            secure: lJ+XbDBgnR3SA1hyBl7kZ+fOJYQYVlZ2A27Vt9p+TNJ6Fsv41Yx7O0fb9bTdiVV+
+
+    matrix:
+        - PYTHON: "C:\\Python35\\python.exe"
+        - PYTHON: "C:\\Python35-x64\\python.exe"
+
+install:
+    - "%PYTHON% -m pip install --upgrade pip wheel setuptools"
+    - "%PYTHON% -m pip install -r .ci/requirements-win.txt"
+
+build_script:
+    - "%PYTHON% setup.py build_ext --inplace"
+
+test_script:
+    - "%PYTHON% -m unittest discover -s tests"
+
+after_test:
+    - "%PYTHON% setup.py bdist_wheel"
+
+artifacts:
+    - path: dist\*
+
+deploy_script:
+    - cmd: "%PYTHON% -m twine upload dist\\*.whl"
+      on:
+        branch: master
+        appveyor_repo_tag: true

.ci/requirements-win.txt

@@ -0,0 +1,2 @@
+cython>=0.24
+twine

asyncpg/cluster.py

@@ -23,6 +23,16 @@
 
 _system = platform.uname().system
 
+if _system == 'Windows':
+    def platform_exe(name):
+        if name.endswith('.exe'):
+            return name
+        return name + '.exe'
+else:
+    def platform_exe(name):
+        return name
+
+
 if _system == 'Linux':
     def ensure_dead_with_parent():
         import ctypes
@@ -165,13 +175,29 @@ def start(self, wait=60, *, server_settings={}, **opts):
         for k, v in server_settings.items():
             extra_args.extend(['-c', '{}={}'.format(k, v)])
 
-        self._daemon_process = \
-            subprocess.Popen(
-                [self._postgres, '-D', self._data_dir, *extra_args],
-                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                preexec_fn=ensure_dead_with_parent)
+        if _system == 'Windows':
+            # On Windows we have to use pg_ctl as direct execution
+            # of postgres daemon under an Administrative account
+            # is not permitted and there is no easy way to drop
+            # privileges.
+            process = subprocess.run(
+                [self._pg_ctl, 'start', '-D', self._data_dir,
+                 '-o', ' '.join(extra_args)],
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            stderr = process.stderr
+
+            if process.returncode != 0:
+                raise ClusterError(
+                    'pg_ctl start exited with status {:d}: {}'.format(
+                        process.returncode, stderr.decode()))
+        else:
+            self._daemon_process = \
+                subprocess.Popen(
+                    [self._postgres, '-D', self._data_dir, *extra_args],
+                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                    preexec_fn=ensure_dead_with_parent)
 
-        self._daemon_pid = self._daemon_process.pid
+            self._daemon_pid = self._daemon_process.pid
 
         self._test_connection(timeout=wait)
 
@@ -291,11 +317,16 @@ def add_hba_entry(self, *, type='host', database, user, address=None,
 
     def trust_local_connections(self):
         self.reset_hba()
-        self.add_hba_entry(type='local', database='all',
-                           user='all', auth_method='trust')
+
+        if _system != 'Windows':
+            self.add_hba_entry(type='local', database='all',
+                               user='all', auth_method='trust')
         self.add_hba_entry(type='host', address='127.0.0.1/32',
                            database='all', user='all',
                            auth_method='trust')
+        self.add_hba_entry(type='host', address='::1/128',
+                           database='all', user='all',
+                           auth_method='trust')
         status = self.get_status()
         if status == 'running':
             self.reload()
@@ -367,8 +398,9 @@ def _test_connection(self, timeout=60):
 
                 try:
                     con = loop.run_until_complete(
-                        asyncpg.connect(database='postgres', timeout=5,
-                                        loop=loop, **self._connection_addr))
+                        asyncpg.connect(database='postgres',
+                                        timeout=5, loop=loop,
+                                        **self._connection_addr))
                 except (OSError, asyncio.TimeoutError,
                         asyncpg.CannotConnectNowError,
                         asyncpg.PostgresConnectionError):
@@ -409,11 +441,13 @@ def _find_pg_config(self, pg_config_path):
         if pg_config_path is None:
             pg_install = os.environ.get('PGINSTALLATION')
             if pg_install:
-                pg_config_path = os.path.join(pg_install, 'pg_config')
+                pg_config_path = platform_exe(
+                    os.path.join(pg_install, 'pg_config'))
             else:
                 pathenv = os.environ.get('PATH').split(os.pathsep)
                 for path in pathenv:
-                    pg_config_path = os.path.join(path, 'pg_config')
+                    pg_config_path = platform_exe(
+                        os.path.join(path, 'pg_config'))
                     if os.path.exists(pg_config_path):
                         break
                 else:
@@ -434,7 +468,7 @@ def _find_pg_binary(self, binary):
                 'could not find {} executable: '.format(binary) +
                 'pg_config output did not provide the BINDIR value')
 
-        bpath = os.path.join(bindir, binary)
+        bpath = platform_exe(os.path.join(bindir, binary))
 
         if not os.path.isfile(bpath):
             raise ClusterError(

asyncpg/connection.py

@@ -410,6 +410,10 @@ async def cancel():
             try:
                 w.write(msg)
                 await r.read()  # Wait until EOF
+            except ConnectionResetError:
+                # On some systems Postgres will reset the connection
+                # after processing the cancellation command.
+                pass
             except Exception as ex:
                 waiter.set_exception(ex)
             finally:

asyncpg/protocol/coreproto.pyx

@@ -645,7 +645,9 @@ cdef class CoreProtocol:
         self.transport = transport
 
         sock = transport.get_extra_info('socket')
-        if sock is not None and sock.family != socket.AF_UNIX:
+        if (sock is not None and
+              (not hasattr(socket, 'AF_UNIX')
+               or sock.family != socket.AF_UNIX)):
             sock.setsockopt(socket.IPPROTO_TCP,
                             socket.TCP_NODELAY, 1)
 

asyncpg/protocol/hton.pxd

@@ -8,11 +8,18 @@
 from libc.stdint cimport int16_t, int32_t, uint16_t, uint32_t, int64_t, uint64_t
 
 
-cdef extern from "arpa/inet.h":
-    uint32_t htonl(uint32_t hostlong)
-    uint16_t htons(uint16_t hostshort)
-    uint32_t ntohl(uint32_t netlong)
-    uint16_t ntohs(uint16_t netshort)
+IF UNAME_SYSNAME == "Windows":
+    cdef extern from "winsock2.h":
+        uint32_t htonl(uint32_t hostlong)
+        uint16_t htons(uint16_t hostshort)
+        uint32_t ntohl(uint32_t netlong)
+        uint16_t ntohs(uint16_t netshort)
+ELSE:
+    cdef extern from "arpa/inet.h":
+        uint32_t htonl(uint32_t hostlong)
+        uint16_t htons(uint16_t hostshort)
+        uint32_t ntohl(uint32_t netlong)
+        uint16_t ntohs(uint16_t netshort)
 
 
 cdef inline void pack_int16(char* buf, int16_t x):

setup.py

@@ -7,6 +7,7 @@
 
 import os
 import os.path
+import platform
 import re
 import sys
 
@@ -21,6 +22,9 @@
 CFLAGS = ['-O2']
 LDFLAGS = []
 
+if platform.uname().system == 'Windows':
+    LDFLAGS.append('ws2_32.lib')
+
 
 class build_ext(_build_ext.build_ext):
     user_options = _build_ext.build_ext.user_options + [

tests/test_connect.py

@@ -8,12 +8,15 @@
 import contextlib
 import ipaddress
 import os
+import platform
 import unittest
 
 import asyncpg
 from asyncpg import _testbase as tb
 from asyncpg.connection import _parse_connect_params
 
+_system = platform.uname().system
+
 
 class TestSettings(tb.ConnectedTestCase):
 
@@ -47,13 +50,20 @@ def setUp(self):
                     ' PASSWORD {!r}'.format(password) if password else ''
                 )
             )
+
+            if _system != 'Windows':
+                self.cluster.add_hba_entry(
+                    type='local',
+                    database='postgres', user='{}_user'.format(method),
+                    auth_method=method)
+
             self.cluster.add_hba_entry(
-                type='local', address=ipaddress.ip_network('127.0.0.0/24'),
+                type='host', address=ipaddress.ip_network('127.0.0.0/24'),
                 database='postgres', user='{}_user'.format(method),
                 auth_method=method)
 
             self.cluster.add_hba_entry(
-                type='host', address=ipaddress.ip_network('127.0.0.0/24'),
+                type='host', address=ipaddress.ip_network('::1/128'),
                 database='postgres', user='{}_user'.format(method),
                 auth_method=method)
 

tests/test_pool.py

@@ -6,10 +6,14 @@
 
 
 import asyncio
+import platform
 
 from asyncpg import _testbase as tb
 
 
+_system = platform.uname().system
+
+
 class TestPool(tb.ConnectedTestCase):
 
     async def test_pool_01(self):
@@ -101,13 +105,19 @@ async def test_pool_auth(self):
 
         self.cluster.reset_hba()
 
+        if _system != 'Windows':
+            self.cluster.add_hba_entry(
+                type='local',
+                database='postgres', user='pooluser',
+                auth_method='md5')
+
         self.cluster.add_hba_entry(
-            type='local',
+            type='host', address='127.0.0.1/32',
             database='postgres', user='pooluser',
             auth_method='md5')
 
         self.cluster.add_hba_entry(
-            type='host', address='127.0.0.0/32',
+            type='host', address='::1/128',
             database='postgres', user='pooluser',
             auth_method='md5')