Add compatibility with PostgreSQL server versions 9.1-9.3

Author: elprans

asyncpg/_testbase.py

@@ -157,6 +157,7 @@ def setUp(self):
         opts = self.getExtraConnectOptions()
         self.con = self.loop.run_until_complete(
             self.cluster.connect(database='postgres', loop=self.loop, **opts))
+        self.server_version = self.con.get_server_version()
 
     def tearDown(self):
         try:

asyncpg/cluster.py

@@ -20,6 +20,7 @@
 import time
 
 import asyncpg
+from asyncpg import serverversion
 
 
 _system = platform.uname().system
@@ -171,8 +172,18 @@ def start(self, wait=60, *, server_settings={}, **opts):
         extra_args = ['--{}={}'.format(k, v) for k, v in opts.items()]
         extra_args.append('--port={}'.format(port))
 
-        if 'unix_socket_directories' not in server_settings:
-            server_settings['unix_socket_directories'] = '/tmp'
+        sockdir = server_settings.get('unix_socket_directories')
+        if sockdir is None:
+            sockdir = server_settings.get('unix_socket_directory')
+        if sockdir is None:
+            sockdir = '/tmp'
+
+        if self._pg_version < (9, 3):
+            sockdir_opt = 'unix_socket_directory'
+        else:
+            sockdir_opt = 'unix_socket_directories'
+
+        server_settings[sockdir_opt] = sockdir
 
         for k, v in server_settings.items():
             extra_args.extend(['-c', '{}={}'.format(k, v)])
@@ -350,6 +361,7 @@ def trust_local_replication_by(self, user):
     def _init_env(self):
         self._pg_config = self._find_pg_config(self._pg_config_path)
         self._pg_config_data = self._run_pg_config(self._pg_config)
+        self._pg_version = self._get_pg_version()
         self._pg_ctl = self._find_pg_binary('pg_ctl')
         self._postgres = self._find_pg_binary('postgres')
 
@@ -493,6 +505,13 @@ def _find_pg_binary(self, binary):
 
         return bpath
 
+    def _get_pg_version(self):
+        version_string = self._pg_config_data.get('version')
+        if not version_string:
+            raise ClusterError('could not determine PostgreSQL version')
+
+        return serverversion.split_server_version_string(version_string)
+
 
 class TempCluster(Cluster):
     def __init__(self, *,

asyncpg/connection.py

@@ -17,6 +17,7 @@
 from . import introspection
 from . import prepared_stmt
 from . import protocol
+from . import serverversion
 from . import transaction
 
 
@@ -29,7 +30,8 @@ class Connection:
     __slots__ = ('_protocol', '_transport', '_loop', '_types_stmt',
                  '_type_by_name_stmt', '_top_xact', '_uid', '_aborted',
                  '_stmt_cache_max_size', '_stmt_cache', '_stmts_to_close',
-                 '_addr', '_opts', '_command_timeout', '_listeners')
+                 '_addr', '_opts', '_command_timeout', '_listeners',
+                 '_server_version', '_intro_query')
 
     def __init__(self, protocol, transport, loop, addr, opts, *,
                  statement_cache_size, command_timeout):
@@ -53,6 +55,15 @@ def __init__(self, protocol, transport, loop, addr, opts, *,
 
         self._listeners = {}
 
+        ver_string = self._protocol.get_settings().server_version
+        self._server_version = \
+            serverversion.split_server_version_string(ver_string)
+
+        if self._server_version < (9, 2):
+            self._intro_query = introspection.INTRO_LOOKUP_TYPES_91
+        else:
+            self._intro_query = introspection.INTRO_LOOKUP_TYPES
+
     async def add_listener(self, channel, callback):
         """Add a listener for Postgres notifications.
 
@@ -84,6 +95,21 @@ def get_server_pid(self):
         """Return the PID of the Postgres server the connection is bound to."""
         return self._protocol.get_server_pid()
 
+    def get_server_version(self):
+        """Return the version of the connected PostgreSQL server.
+
+        The returned value is a named tuple similar to that in
+        ``sys.version_info``:
+
+        .. code-block:: pycon
+
+            >>> con.get_server_version()
+            ServerVersion(major=9, minor=6, micro=1,
+                          releaselevel='final', serial=0)
+
+        """
+        return self._server_version
+
     def get_settings(self):
         """Return connection settings.
 
@@ -188,8 +214,7 @@ async def _get_statement(self, query, timeout):
         ready = state._init_types()
         if ready is not True:
             if self._types_stmt is None:
-                self._types_stmt = await self.prepare(
-                    introspection.INTRO_LOOKUP_TYPES)
+                self._types_stmt = await self.prepare(self._intro_query)
 
             types = await self._types_stmt.fetch(list(ready))
             protocol.get_settings().register_data_types(types)

asyncpg/introspection.py

@@ -123,6 +123,122 @@
 '''
 
 
+# Prior to 9.2 PostgreSQL did not have range types.
+INTRO_LOOKUP_TYPES_91 = '''\
+WITH RECURSIVE typeinfo_tree(
+    oid, ns, name, kind, basetype, elemtype, range_subtype,
+    elem_has_bin_input, elem_has_bin_output, attrtypoids, attrnames, depth)
+AS (
+    WITH composite_attrs
+    AS (
+        SELECT
+            c.reltype                                        AS comptype_oid,
+            array_agg(ia.atttypid ORDER BY ia.attnum)        AS typoids,
+            array_agg(ia.attname::text ORDER BY ia.attnum)   AS names
+        FROM
+            pg_attribute ia
+            INNER JOIN pg_class c
+                ON (ia.attrelid = c.oid)
+        WHERE
+            ia.attnum > 0 AND NOT ia.attisdropped
+        GROUP BY
+            c.reltype
+    ),
+
+    typeinfo
+    AS (
+        SELECT
+            t.oid                           AS oid,
+            ns.nspname                      AS ns,
+            t.typname                       AS name,
+            t.typtype                       AS kind,
+            (CASE WHEN t.typtype = 'd' THEN
+                (WITH RECURSIVE typebases(oid, depth) AS (
+                    SELECT
+                        t2.typbasetype      AS oid,
+                        0                   AS depth
+                    FROM
+                        pg_type t2
+                    WHERE
+                        t2.oid = t.oid
+
+                    UNION ALL
+
+                    SELECT
+                        t2.typbasetype      AS oid,
+                        tb.depth + 1        AS depth
+                    FROM
+                        pg_type t2,
+                        typebases tb
+                    WHERE
+                       tb.oid = t2.oid
+                       AND t2.typbasetype != 0
+               ) SELECT oid FROM typebases ORDER BY depth DESC LIMIT 1)
+
+               ELSE NULL
+            END)                            AS basetype,
+            t.typelem                       AS elemtype,
+            NULL::oid                       AS range_subtype,
+            elem_t.typreceive::oid != 0     AS elem_has_bin_input,
+            elem_t.typsend::oid != 0        AS elem_has_bin_output,
+            (CASE WHEN t.typtype = 'c' THEN
+                (SELECT ca.typoids
+                FROM composite_attrs AS ca
+                WHERE ca.comptype_oid = t.oid)
+
+                ELSE NULL
+            END)                            AS attrtypoids,
+            (CASE WHEN t.typtype = 'c' THEN
+                (SELECT ca.names
+                FROM composite_attrs AS ca
+                WHERE ca.comptype_oid = t.oid)
+
+                ELSE NULL
+            END)                            AS attrnames
+        FROM
+            pg_catalog.pg_type AS t
+            INNER JOIN pg_catalog.pg_namespace ns ON (
+                ns.oid = t.typnamespace)
+            LEFT JOIN pg_type elem_t ON (
+                t.typlen = -1 AND
+                t.typelem != 0 AND
+                t.typelem = elem_t.oid
+            )
+    )
+
+    SELECT
+        ti.oid, ti.ns, ti.name, ti.kind, ti.basetype, ti.elemtype,
+        ti.range_subtype, ti.elem_has_bin_input, ti.elem_has_bin_output,
+        ti.attrtypoids, ti.attrnames, 0
+    FROM
+        typeinfo AS ti
+    WHERE
+        ti.oid = any($1::oid[])
+
+    UNION ALL
+
+    SELECT
+        ti.oid, ti.ns, ti.name, ti.kind, ti.basetype, ti.elemtype,
+        ti.range_subtype, ti.elem_has_bin_input, ti.elem_has_bin_output,
+        ti.attrtypoids, ti.attrnames, tt.depth + 1
+    FROM
+        typeinfo ti,
+        typeinfo_tree tt
+    WHERE
+        (tt.elemtype IS NOT NULL AND ti.oid = tt.elemtype)
+        OR (tt.attrtypoids IS NOT NULL AND ti.oid = any(tt.attrtypoids))
+        OR (tt.range_subtype IS NOT NULL AND ti.oid = tt.range_subtype)
+)
+
+SELECT DISTINCT
+    *
+FROM
+    typeinfo_tree
+ORDER BY
+    depth DESC
+'''
+
+
 TYPE_BY_NAME = '''\
 SELECT
     t.oid,

asyncpg/serverversion.py

@@ -0,0 +1,33 @@
+# Copyright (C) 2016-present the ayncpg authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of asyncpg and is released under
+# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0
+
+
+from . import types
+
+
+def split_server_version_string(version_string):
+    version_string = version_string.strip()
+    if version_string.startswith('PostgreSQL '):
+        version_string = version_string[len('PostgreSQL '):]
+
+    parts = version_string.strip().split('.')
+    if not parts[-1].isdigit():
+        # release level specified
+        level = parts[-1].rstrip('0123456789').lower()
+        serial = parts[-1][level:]
+        versions = [int(p) for p in parts[:-1]][:3]
+    else:
+        level = 'final'
+        serial = 0
+        versions = [int(p) for p in parts][:3]
+
+    if len(versions) < 3:
+        versions += [0] * (3 - len(versions))
+
+    versions.append(level)
+    versions.append(serial)
+
+    return types.ServerVersion(*versions)

asyncpg/types.py

@@ -10,7 +10,7 @@
 
 __all__ = (
     'Type', 'Attribute', 'Range', 'BitString', 'Point', 'Path', 'Polygon',
-    'Box', 'Line', 'LineSegment', 'Circle'
+    'Box', 'Line', 'LineSegment', 'Circle', 'ServerVersion'
 )
 
 
@@ -29,6 +29,11 @@
 Attribute.type.__doc__ = 'Attribute data type :class:`asyncpg.types.Type`.'
 
 
+ServerVersion = collections.namedtuple(
+    'ServerVersion', ['major', 'minor', 'micro', 'releaselevel', 'serial'])
+ServerVersion.__doc__ = 'PostgreSQL server version tuple.'
+
+
 class Range:
     """Immutable representation of PostgreSQL `range` type."""
 

tests/test_codecs.py

@@ -11,6 +11,7 @@
 import math
 import random
 import struct
+import unittest
 import uuid
 
 import asyncpg
@@ -209,11 +210,11 @@ def _timezone(offset):
     ('json', 'json', [
         '[1, 2, 3, 4]',
         '{"a": [1, 2], "b": 0}'
-    ]),
+    ], (9, 2)),
     ('jsonb', 'jsonb', [
         '[1, 2, 3, 4]',
         '{"a": [1, 2], "b": 0}'
-    ]),
+    ], (9, 4)),
     ('oid[]', 'oid[]', [
         [1, 2, 3, 4],
         []
@@ -333,7 +334,7 @@ def _timezone(offset):
     ]),
     ('line', 'line', [
         asyncpg.Line(1, 2, 3),
-    ]),
+    ], (9, 4)),
     ('lseg', 'lseg', [
         asyncpg.LineSegment((1, 2), (2, 2)),
     ]),
@@ -351,7 +352,10 @@ class TestCodecs(tb.ConnectedTestCase):
 
     async def test_standard_codecs(self):
         """Test encoding/decoding of standard data types and arrays thereof."""
-        for (typname, intname, sample_data) in type_samples:
+        for (typname, intname, sample_data, *metadata) in type_samples:
+            if metadata and self.server_version < metadata[0]:
+                continue
+
             st = await self.con.prepare(
                 "SELECT $1::" + typname
             )
@@ -548,7 +552,8 @@ async def test_arrays(self):
                 "SELECT $1::int[]",
                 [[1], ['t'], [2]])
 
-        with self.assertRaisesRegex(TypeError, 'non-trivial iterable expected'):
+        with self.assertRaisesRegex(TypeError,
+                                    'non-trivial iterable expected'):
             await self.con.fetchval(
                 "SELECT $1::int[]",
                 1)
@@ -641,6 +646,11 @@ async def test_domains(self):
 
     async def test_range_types(self):
         """Test encoding/decoding of range types."""
+
+        if self.server_version < (9, 2):
+            raise unittest.SkipTest(
+                'PostgreSQL servers < 9.2 do not support range types.')
+
         cases = [
             ('int4range', [
                 [(1, 9), asyncpg.Range(1, 10)],

tests/test_prepare.py

@@ -8,6 +8,7 @@
 import asyncio
 import asyncpg
 import gc
+import unittest
 
 from asyncpg import _testbase as tb
 
@@ -397,6 +398,11 @@ async def test_prepare_21_errors(self):
         self.assertEqual(await stmt.fetchval(5), 2)
 
     async def test_prepare_22_empty(self):
+        # Support for empty target list was added in PostgreSQL 9.4
+        if self.server_version < (9, 4):
+            raise unittest.SkipTest(
+                'PostgreSQL servers < 9.4 do not support empty target list.')
+
         result = await self.con.fetchrow('SELECT')
         self.assertEqual(result, ())
         self.assertEqual(repr(result), '<Record>')