- genericize the test for ischema_names

- some fixes to the patch to handle empty args, whitespace
- changelog clarifies where this API fits at the moment

diff --git a/CHANGES b/CHANGES
index 00b60d510bf40b70b93b62ded970da68bb24145e..cea3cda2650386565a331e0b4a050b4ee5a45c32 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -633,6 +633,30 @@ underneath "0.7.xx".
     The phrase is established using with_hint().
     Courtesy Ryan Kelly [ticket:2506]
 
+  - [feature] The "ischema_names" dictionary of the
+    Postgresql dialect is "unofficially" customizable.
+    Meaning, new types such as PostGIS types can
+    be added into this dictionary, and the PG type
+    reflection code should be able to handle simple
+    types with variable numbers of arguments.
+    The functionality here is "unofficial" for
+    three reasons:
+
+    1. this is not an "official" API.  Ideally
+       an "official" API would allow custom type-handling
+       callables at the dialect or global level
+       in a generic way.
+    2. This is only implemented for the PG dialect,
+       in particular because PG has broad support
+       for custom types vs. other database backends.
+       A real API would be implemented at the
+       default dialect level.
+    3. The reflection code here is only tested against
+       simple types and probably has issues with more
+       compositional types.
+
+    patch courtesy Éric Lemoine.
+
 - firebird
   - [feature] The "startswith()" operator renders
     as "STARTING WITH", "~startswith()" renders

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index f59aa50b4de1d574d4d61d199862e4dd8fe164e7..4e30a76c4f0276d8046a80ad800eda993c51e5df 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1471,7 +1471,7 @@ class PGDialect(default.DefaultDialect):
         """
         s = sql.text(SQL_COLS,
             bindparams=[sql.bindparam('table_oid', type_=sqltypes.Integer)],
-            typemap={'attname':sqltypes.Unicode, 'default':sqltypes.Unicode}
+            typemap={'attname': sqltypes.Unicode, 'default': sqltypes.Unicode}
         )
         c = connection.execute(s, table_oid=table_oid)
         rows = c.fetchall()
@@ -1501,8 +1501,8 @@ class PGDialect(default.DefaultDialect):
         if charlen:
             charlen = charlen.group(1)
         args = re.search('\((.*)\)', format_type)
-        if args:
-            args = tuple(args.group(1).split(','))
+        if args and args.group(1):
+            args = tuple(re.split('\s*,\s*', args.group(1)))
         else:
             args = ()
         kwargs = {}
@@ -1535,7 +1535,7 @@ class PGDialect(default.DefaultDialect):
                 args = (int(charlen),)
             else:
                 args = ()
-        elif attype in ('interval','interval year to month',
+        elif attype in ('interval', 'interval year to month',
                             'interval day to second'):
             if charlen:
                 kwargs['precision'] = int(charlen)

diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py
index 44955c9b80da71055ff6550c2bfe223f6e5b5dd3..0bd6666e2293f490559ac3f170f9e75ed8cfea8f 100644 (file)
--- a/test/dialect/test_postgresql.py
+++ b/test/dialect/test_postgresql.py
@@ -1713,54 +1713,49 @@ class ReflectionTest(fixtures.TestBase):
         eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}])
         conn.close()
 
-class PostGISColumnReflection(fixtures.TestBase):
-    __only_on__ = 'postgresql'
+class CustomTypeReflectionTest(fixtures.TestBase):
 
-    class Geometry(object):
-        def __init__(self, geometry_type=None, srid=None):
-            self.geometry_type = geometry_type
-            self.srid = srid
+    class CustomType(object):
+        def __init__(self, arg1=None, arg2=None):
+            self.arg1 = arg1
+            self.arg2 = arg2
 
     ischema_names = None
 
-    @classmethod
-    def setup_class(cls):
+    def setup(self):
         ischema_names = postgresql.PGDialect.ischema_names
         postgresql.PGDialect.ischema_names = ischema_names.copy()
-        postgresql.PGDialect.ischema_names['geometry'] = cls.Geometry
-        cls.ischema_names = ischema_names
-
-    @classmethod
-    def teardown_class(cls):
-        postgresql.PGDialect.ischema_names = cls.ischema_names
-        cls.ischema_names = None
+        self.ischema_names = ischema_names
 
-    def test_geometry(self):
-        dialect = postgresql.PGDialect()
-        column_info = dialect._get_column_info(
-                'geom', 'geometry', None, False,
+    def teardown(self):
+        postgresql.PGDialect.ischema_names = self.ischema_names
+        self.ischema_names = None
+
+    def _assert_reflected(self, dialect):
+        for sch, args in [
+            ('my_custom_type', (None, None)),
+            ('my_custom_type()', (None, None)),
+            ('my_custom_type(ARG1)', ('ARG1', None)),
+            ('my_custom_type(ARG1, ARG2)', ('ARG1', 'ARG2')),
+        ]:
+            column_info = dialect._get_column_info(
+                'colname', sch, None, False,
                 {}, {}, 'public')
-        assert isinstance(column_info['type'], self.Geometry)
-        assert column_info['type'].geometry_type is None
-        assert column_info['type'].srid is None
+            assert isinstance(column_info['type'], self.CustomType)
+            eq_(column_info['type'].arg1, args[0])
+            eq_(column_info['type'].arg2, args[1])
 
-    def test_geometry_with_type(self):
+    def test_clslevel(self):
+        postgresql.PGDialect.ischema_names['my_custom_type'] = self.CustomType
         dialect = postgresql.PGDialect()
-        column_info = dialect._get_column_info(
-                'geom', 'geometry(POLYGON)', None, False,
-                {}, {}, 'public')
-        assert isinstance(column_info['type'], self.Geometry)
-        assert column_info['type'].geometry_type == 'POLYGON'
-        assert column_info['type'].srid is None
+        self._assert_reflected(dialect)
 
-    def test_geometry_with_type_and_srid(self):
+    def test_instancelevel(self):
         dialect = postgresql.PGDialect()
-        column_info = dialect._get_column_info(
-                'geom', 'geometry(POLYGON,4326)', None, False,
-                {}, {}, 'public')
-        assert isinstance(column_info['type'], self.Geometry)
-        assert column_info['type'].geometry_type == 'POLYGON'
-        assert column_info['type'].srid == '4326'
+        dialect.ischema_names = dialect.ischema_names.copy()
+        dialect.ischema_names['my_custom_type'] = self.CustomType
+        self._assert_reflected(dialect)
+
 
 class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):