current_schema = schema
else:
current_schema = self.default_schema_name
- s = """
- SELECT definition FROM pg_views
- WHERE schemaname = :schema
- AND viewname = :view_name
- """
+
+ if self.server_version_info >= (9, 3):
+ s = """
+ SELECT definition FROM pg_views
+ WHERE schemaname = :schema
+ AND viewname = :view_name
+
+ UNION
+
+ SELECT definition FROM pg_matviews
+ WHERE schemaname = :schema
+ AND matviewname = :view_name
+
+ """
+ else:
+ s = """
+ SELECT definition FROM pg_views
+ WHERE schemaname = :schema
+ AND viewname = :view_name
+ """
+
rp = connection.execute(sql.text(s),
view_name=view_name, schema=current_schema)
if rp:
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import base as postgresql
from sqlalchemy.dialects.postgresql import ARRAY
+import re
class ForeignTableReflectionTest(fixtures.TablesTest, AssertsExecutionResults):
insp = inspect(testing.db)
eq_(set(insp.get_view_names()), set(['test_mview', 'test_regview']))
+ def test_get_view_definition(self):
+ insp = inspect(testing.db)
+ eq_(
+ re.sub(
+ r'[\n\t ]+', ' ',
+ insp.get_view_definition("test_mview").strip()),
+ "SELECT testtable.id, testtable.data FROM testtable;"
+ )
+
class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
"""Test PostgreSQL domains"""