})
]
- reflection_options = ('postgresql_ignore_search_path', 'postgresql_relkind')
+ reflection_options = ('postgresql_ignore_search_path',)
_backslash_escapes = True
return tuple([int(x) for x in m.group(1, 2, 3) if x is not None])
@reflection.cache
- def get_table_oid(self, connection, table_name, schema=None, postgresql_relkind=None, **kw):
+ def get_table_oid(self, connection, table_name, schema=None, **kw):
"""Fetch the oid for schema.table_name.
Several reflection methods require the table oid. The idea for using
else:
schema_where_clause = "pg_catalog.pg_table_is_visible(c.oid)"
- RELKIND_SYNONYMS = {
- 'materialized': 'm',
- 'foreign': 'f',
- 'view': 'v'
- }
- ACCEPTED_RELKINDS = ('r','v','m','f')
- if postgresql_relkind is None:
- postgresql_relkind = 'r'
- else:
- postgresql_relkind = postgresql_relkind.lower()
- if postgresql_relkind in RELKIND_SYNONYMS:
- postgresql_relkind = RELKIND_SYNONYMS[postgresql_relkind.lower()]
- if postgresql_relkind not in ACCEPTED_RELKINDS:
- raise exc.SQLAlchemyError('Invalid postgresql_relkind: %s' % postgresql_relkind)
-
query = """
SELECT c.oid
FROM pg_catalog.pg_class c
LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
WHERE (%s)
- AND c.relname = :table_name AND c.relkind in ('%s', 'v')
- """ % (schema_where_clause, postgresql_relkind)
+ AND c.relname = :table_name AND c.relkind in ('r', 'v', 'm', 'f')
+ """ % schema_where_clause
# Since we're binding to unicode, table_name and schema_name must be
# unicode.
table_name = util.text_type(table_name)
def get_columns(self, connection, table_name, schema=None, **kw):
table_oid = self.get_table_oid(connection, table_name, schema,
- info_cache=kw.get('info_cache'),
- postgresql_relkind=kw.get('postgresql_relkind'))
+ info_cache=kw.get('info_cache'))
SQL_COLS = """
SELECT a.attname,
pg_catalog.format_type(a.atttypid, a.atttypmod),
@reflection.cache
def get_pk_constraint(self, connection, table_name, schema=None, **kw):
table_oid = self.get_table_oid(connection, table_name, schema,
- info_cache=kw.get('info_cache'),
- postgresql_relkind=kw.get('postgresql_relkind'))
+ info_cache=kw.get('info_cache'))
if self.server_version_info < (8, 4):
PK_SQL = """
postgresql_ignore_search_path=False, **kw):
preparer = self.identifier_preparer
table_oid = self.get_table_oid(connection, table_name, schema,
- info_cache=kw.get('info_cache'),
- postgresql_relkind=kw.get('postgresql_relkind'))
+ info_cache=kw.get('info_cache'))
FK_SQL = """
SELECT r.conname,
@reflection.cache
def get_indexes(self, connection, table_name, schema, **kw):
table_oid = self.get_table_oid(connection, table_name, schema,
- info_cache=kw.get('info_cache'),
- postgresql_relkind=kw.get('postgresql_relkind'))
+ info_cache=kw.get('info_cache'))
# cast indkey as varchar since it's an int2vector,
# returned as a list by some drivers such as pypostgresql
def get_unique_constraints(self, connection, table_name,
schema=None, **kw):
table_oid = self.get_table_oid(connection, table_name, schema,
- info_cache=kw.get('info_cache'),
- postgresql_relkind=kw.get('postgresql_relkind'))
+ info_cache=kw.get('info_cache'))
UNIQUE_SQL = """
SELECT
from sqlalchemy.dialects.postgresql import base as postgresql
-class RelKindReflectionTest(fixtures.TestBase, AssertsExecutionResults):
- """Test postgresql_relkind reflection option"""
+class AlternateRelkindReflectionTest(fixtures.TestBase, AssertsExecutionResults):
+ """Test reflection on materialized views and foreign tables"""
__requires__ = 'postgresql_test_dblink',
__only_on__ = 'postgresql >= 9.3'
con.execute('DROP TABLE testtable;')
def test_mview_is_reflected(self):
- mview_relkind_names = ('m', 'materialized')
- for mview_relkind_name in mview_relkind_names:
- metadata = MetaData(testing.db)
- table = Table('test_mview', metadata, autoload=True, postgresql_relkind=mview_relkind_name)
- eq_(set(table.columns.keys()), set(['id', 'data']), "Columns of reflected mview didn't equal expected columns")
+ metadata = MetaData(testing.db)
+ table = Table('test_mview', metadata, autoload=True)
+ eq_(set(table.columns.keys()), set(['id', 'data']), "Columns of reflected mview didn't equal expected columns")
def test_mview_select(self):
metadata = MetaData(testing.db)
- table = Table('test_mview', metadata, autoload=True, postgresql_relkind='m')
+ table = Table('test_mview', metadata, autoload=True)
assert table.select().execute().fetchall() == [
(89, 'd1',)
]
def test_foreign_table_is_reflected(self):
- foreign_table_relkind_names = ('f', 'foreign')
- for foreign_table_relkind_name in foreign_table_relkind_names:
- metadata = MetaData(testing.db)
- table = Table('test_foreigntable', metadata, autoload=True, postgresql_relkind=foreign_table_relkind_name)
- eq_(set(table.columns.keys()), set(['id', 'data']), "Columns of reflected foreign table didn't equal expected columns")
+ metadata = MetaData(testing.db)
+ table = Table('test_foreigntable', metadata, autoload=True)
+ eq_(set(table.columns.keys()), set(['id', 'data']), "Columns of reflected foreign table didn't equal expected columns")
def test_foreign_table_select(self):
metadata = MetaData(testing.db)
- table = Table('test_foreigntable', metadata, autoload=True, postgresql_relkind='f')
+ table = Table('test_foreigntable', metadata, autoload=True)
assert table.select().execute().fetchall() == [
(89, 'd1',)
]
def test_foreign_table_roundtrip(self):
metadata = MetaData(testing.db)
- table = Table('test_foreigntable', metadata, autoload=True, postgresql_relkind='f')
+ table = Table('test_foreigntable', metadata, autoload=True)
connection = testing.db.connect()
trans = connection.begin()
(89, 'd1',)
]
- def test_invalid_relkind(self):
- metadata = MetaData(testing.db)
- def create_bad_table():
- return Table('test_foreigntable', metadata, autoload=True, postgresql_relkind='nope')
-
- assert_raises(exc.SQLAlchemyError, create_bad_table)
-
class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
"""Test PostgreSQL domains"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
