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>')