The ``has_table`` method now also checks views

The :meth:`_engine.Inspector.has_table` method will now consistently check
for views of the given name as well as tables. Previously this behavior was
dialect dependent, with PostgreSQL, MySQL/MariaDB and SQLite supporting it,
and Oracle and SQL Server not supporting it. Third party dialects should
also seek to ensure their :meth:`_engine.Inspector.has_table` method
searches for views as well as tables for the given name.

Fixes: #7161
Change-Id: I9e523c76741b19596c81ef577dc6f0823e44183b

6 files changed, 64 insertions, 7 deletions

diff --git a/doc/build/changelog/unreleased_20/7161.rst b/doc/build/changelog/unreleased_20/7161.rst
new file mode 100644
index 000000000..ec584b3a9
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/7161.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: engine, bug
+    :tickets: 7161
+
+    The :meth:`_engine.Inspector.has_table` method will now consistently check
+    for views of the given name as well as tables. Previously this behavior was
+    dialect dependent, with PostgreSQL, MySQL/MariaDB and SQLite supporting it,
+    and Oracle and SQL Server not supporting it. Third party dialects should
+    also seek to ensure their :meth:`_engine.Inspector.has_table` method
+    searches for views as well as tables for the given name.
\ No newline at end of file
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 95ccd2ca8..8c8260f3b 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2888,6 +2888,8 @@ class MSDialect(default.DefaultDialect):
     def has_table(self, connection, tablename, dbname, owner, schema):
         self._ensure_has_table_connection(connection)
         if tablename.startswith("#"):  # temporary table
+            # mssql does not support temporary views
+            # SQL Error [4103] [S0001]: "#v": Temporary views are not allowed
             tables = ischema.mssql_temp_table_columns
 
             s = sql.select(tables.c.table_name).where(
@@ -2915,7 +2917,10 @@ class MSDialect(default.DefaultDialect):
 
             s = sql.select(tables.c.table_name).where(
                 sql.and_(
-                    tables.c.table_type == "BASE TABLE",
+                    sql.or_(
+                        tables.c.table_type == "BASE TABLE",
+                        tables.c.table_type == "VIEW",
+                    ),
                     tables.c.table_name == tablename,
                 )
             )
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 8b790c70c..c0bf985e5 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1611,9 +1611,14 @@ class OracleDialect(default.DefaultDialect):
 
         cursor = connection.execute(
             sql.text(
-                "SELECT table_name FROM all_tables "
-                "WHERE table_name = CAST(:name AS VARCHAR2(128)) "
-                "AND owner = CAST(:schema_name AS VARCHAR2(128))"
+                """SELECT table_name FROM all_tables
+                WHERE table_name = CAST(:name AS VARCHAR2(128))
+                AND owner = CAST(:schema_name AS VARCHAR2(128))
+                UNION ALL
+                SELECT view_name FROM all_views
+                WHERE view_name = CAST(:name AS VARCHAR2(128))
+                AND owner = CAST(:schema_name AS VARCHAR2(128))
+                """
             ),
             dict(
                 name=self.denormalize_name(table_name),
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index d1484718e..fdaeaddcd 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -463,7 +463,7 @@ class Dialect(object):
 
     def has_table(self, connection, table_name, schema=None, **kw):
         """For internal dialect use, check the existence of a particular table
-        in the database.
+        or view in the database.
 
         Given a :class:`_engine.Connection` object, a string table_name and
         optional schema name, return True if the given table exists in the
@@ -481,6 +481,13 @@ class Dialect(object):
            Alternatively, for legacy cross-compatibility, the
            :meth:`_engine.Engine.has_table` method may be used.
 
+        .. versionchanged:: 2.0
+
+            The :meth:`_engine.Dialect.has_table` method should also check
+            for the presence of views.  In previous versions this
+            behavior was dialect specific. New dialect suite tests were added
+            to ensure that dialects conform with this behavior consistently.
+
         """
 
         raise NotImplementedError()
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 113aa8ea0..d8c2bcbd7 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -268,8 +268,7 @@ class Inspector(object):
             )
 
     def has_table(self, table_name, schema=None):
-        """Return True if the backend has a table of the given name.
-
+        """Return True if the backend has a table or view of the given name.
 
         :param table_name: name of the table to check
         :param schema: schema name to query, if not the default schema.
@@ -277,6 +276,11 @@ class Inspector(object):
         .. versionadded:: 1.4 - the :meth:`.Inspector.has_table` method
            replaces the :meth:`_engine.Engine.has_table` method.
 
+        .. versionchanged:: 2.0:: The method checks also for views.
+           In previous version this behaviour was dialect specific. New
+           dialect suite tests were added to ensure all dialect conform with
+           this behaviour.
+
         """
         # TODO: info_cache?
         with self._operation_context() as conn:
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 88189c2d9..6fbd74689 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -77,6 +77,32 @@ class HasTableTest(fixtures.TablesTest):
                 )
             )
 
+    @testing.requires.views
+    def test_has_table_view(self, connection):
+        query = "CREATE VIEW vv AS SELECT * FROM test_table"
+        connection.execute(sa.sql.text(query))
+        insp = inspect(connection)
+        try:
+            is_true(insp.has_table("vv"))
+        finally:
+            connection.execute(sa.sql.text("DROP VIEW vv"))
+
+    @testing.requires.views
+    @testing.requires.schemas
+    def test_has_table_view_schema(self, connection):
+        query = "CREATE VIEW %s.vv AS SELECT * FROM %s.test_table_s" % (
+            config.test_schema,
+            config.test_schema,
+        )
+        connection.execute(sa.sql.text(query))
+        insp = inspect(connection)
+        try:
+            is_true(insp.has_table("vv", config.test_schema))
+        finally:
+            connection.execute(
+                sa.sql.text("DROP VIEW %s.vv" % config.test_schema)
+            )
+
 
 class HasIndexTest(fixtures.TablesTest):
     __backend__ = True