diff options
| -rw-r--r-- | CHANGES | 24 | ||||
| -rw-r--r-- | lib/sqlalchemy/dialects/postgresql/base.py | 8 | ||||
| -rw-r--r-- | test/dialect/test_postgresql.py | 67 |
3 files changed, 59 insertions, 40 deletions
@@ -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 f59aa50b4..4e30a76c4 100644 --- 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 44955c9b8..0bd6666e2 100644 --- 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): |