refactor(api): mysql ddl accessor implementation

ibis/backends/mysql/__init__.py

@@ -24,7 +24,8 @@
 from ibis.backends import CanCreateDatabase
 from ibis.backends.mysql.datatypes import _type_from_cursor_info
 from ibis.backends.sql import SQLBackend
-from ibis.backends.sql.compilers.base import STAR, TRUE, C
+from ibis.backends.sql.compilers.base import STAR, C
+from ibis.util import deprecated
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -289,79 +290,95 @@ def raw_sql(self, query: str | sg.Expression, **kwargs: Any) -> Any:
             con.commit()
             return cursor
 
-    # TODO: disable positional arguments
-    def list_tables(
+    def _list_query_constructor(self, col: str, where_predicates: list) -> str:
+        """Helper function to construct sqlglot queries for _list_* methods."""
+
+        sg_query = (
+            sg.select(col)
+            .from_(sg.table("tables", db="information_schema"))
+            .where(*where_predicates)
+        ).sql(self.name)
+
+        return sg_query
+
+    def _list_objects(
+        self,
+        like: str | None,
+        database: tuple[str, str] | str | None,
+        object_type: str,
+    ) -> list[str]:
+        """Generic method to list objects like tables or views."""
+
+        table_loc = self._warn_and_create_table_loc(database)
+
+        ## having an issue as it seem mysql doesn't have a self.current_catalog
+        ## not clear to me why, my guess is it doesn't support catalogs but unclear
+        # catalog = table_loc.catalog or self.current_catalog
+        database = table_loc.db or self.current_database
+
+        col = "table_name"
+        where_predicates = [
+            # C.table_catalog.eq(sge.convert(catalog)),
+            C.table_schema.eq(sge.convert(database)),
+            C.table_type.eq(object_type),
+        ]
+
+        sql = self._list_query_constructor(col, where_predicates)
+
+        with self._safe_raw_sql(sql) as cur:
+            out = cur.fetchall()
+
+        return self._filter_with_like(map(itemgetter(0), out), like)
+
+    def _list_tables(
         self,
         like: str | None = None,
-        schema: str | None = None,
         database: tuple[str, str] | str | None = None,
     ) -> list[str]:
-        """List the tables in the database.
+        """List physical tables."""
 
-        ::: {.callout-note}
-        ## Ibis does not use the word `schema` to refer to database hierarchy.
+        return self._list_objects(like, database, "BASE TABLE")
 
-        A collection of tables is referred to as a `database`.
-        A collection of `database` is referred to as a `catalog`.
+    def _list_temp_tables(
+        self,
+        like: str | None = None,
+        database: tuple[str, str] | str | None = None,
+    ) -> list[str]:
+        """List temporary tables."""
 
-        These terms are mapped onto the corresponding features in each
-        backend (where available), regardless of whether the backend itself
-        uses the same terminology.
-        :::
+        return self._list_objects(like, database, "TEMPORARY")
 
-        Parameters
-        ----------
-        like
-            A pattern to use for listing tables.
-        schema
-            [deprecated] The schema to perform the list against.
-        database
-            Database to list tables from. Default behavior is to show tables in
-            the current database (`self.current_database`).
-        """
-        if schema is not None:
-            self._warn_schema()
-
-        if schema is not None and database is not None:
-            raise ValueError(
-                "Using both the `schema` and `database` kwargs is not supported. "
-                "`schema` is deprecated and will be removed in Ibis 10.0"
-                "\nUse the `database` kwarg with one of the following patterns:"
-                '\ndatabase="database"'
-                '\ndatabase=("catalog", "database")'
-                '\ndatabase="catalog.database"',
-            )
-        elif schema is not None:
-            table_loc = schema
-        elif database is not None:
-            table_loc = database
-        else:
-            table_loc = self.current_database
+    def _list_views(
+        self,
+        like: str | None = None,
+        database: tuple[str, str] | str | None = None,
+    ) -> list[str]:
+        """List views."""
 
-        table_loc = self._to_sqlglot_table(table_loc)
+        return self._list_objects(like, database, "VIEW")
 
-        conditions = [TRUE]
+    # TODO: disable positional arguments
+    @deprecated(as_of="10.0", instead="use the con.tables")
+    def list_tables(
+        self,
+        like: str | None = None,
+        schema: str | None = None,
+        database: tuple[str, str] | str | None = None,
+    ) -> list[str]:
+        """List the tables and views in the database."""
+        table_loc = self._warn_and_create_table_loc(database, schema)
 
-        if (sg_cat := table_loc.args["catalog"]) is not None:
-            sg_cat.args["quoted"] = False
-        if (sg_db := table_loc.args["db"]) is not None:
-            sg_db.args["quoted"] = False
-        if table_loc.catalog or table_loc.db:
-            conditions = [C.table_schema.eq(sge.convert(table_loc.sql(self.name)))]
+        database = self.current_database
+        if table_loc is not None:
+            database = table_loc.db or database
 
-        col = "table_name"
-        sql = (
-            sg.select(col)
-            .from_(sg.table("tables", db="information_schema"))
-            .distinct()
-            .where(*conditions)
-            .sql(self.name)
+        tables_and_views = list(
+            set(self._list_tables(like=like, database=database))
+            | set(self._list_temp_tables(like=like, database=database))
+            | set(self._list_views(like=like, database=database))
         )
 
-        with self._safe_raw_sql(sql) as cur:
-            out = cur.fetchall()
-
-        return self._filter_with_like(map(itemgetter(0), out), like)
+        return tables_and_views
 
     def execute(
         self, expr: ir.Expr, limit: str | None = "default", **kwargs: Any
@@ -480,7 +497,7 @@ def _register_in_memory_table(self, op: ops.InMemoryTable) -> None:
             )
 
         # only register if we haven't already done so
-        if (name := op.name) not in self.list_tables():
+        if (name := op.name) not in self.tables:
             quoted = self.compiler.quoted
             column_defs = [
                 sg.exp.ColumnDef(

ibis/backends/mysql/tests/test_client.py

@@ -234,13 +234,13 @@ def test_list_tables_schema_warning_refactor(con):
         "event",
         "func",
     }
-    assert con.list_tables()
+    assert con.tables
 
     with pytest.warns(FutureWarning):
         assert mysql_tables.issubset(con.list_tables(schema="mysql"))
 
-    assert mysql_tables.issubset(con.list_tables(database="mysql"))
-    assert mysql_tables.issubset(con.list_tables(database=("mysql",)))
+    assert mysql_tables.issubset(con.ddl.list_tables(database="mysql"))
+    assert mysql_tables.issubset(con.ddl.list_tables(database=("mysql",)))
 
 
 def test_invalid_port():

ibis/backends/tests/test_api.py

@@ -157,7 +157,10 @@ def test_list_tables(ddl_con):
 
     assert table_name not in ddl_con.ddl.list_views()
     assert table_name not in ddl_con.ddl.list_temp_tables()
-    assert table_name not in ddl_con.ddl.list_temp_views()
+    try:
+        assert table_name not in ddl_con.ddl.list_temp_views()
+    except NotImplementedError:  # not all backends have list_temp_views
+        return
 
 
 def test_list_views(ddl_con, temp_view):
@@ -171,7 +174,10 @@ def test_list_views(ddl_con, temp_view):
     assert temp_view in views
     assert temp_view not in ddl_con.ddl.list_tables()
     assert temp_view not in ddl_con.ddl.list_temp_tables()
-    assert temp_view not in ddl_con.ddl.list_temp_views()
+    try:
+        assert temp_view not in ddl_con.ddl.list_temp_views()
+    except NotImplementedError:  # not all backends have list_temp_views
+        return
 
 
 def test_list_temp_tables(ddl_con):
@@ -183,10 +189,14 @@ def test_list_temp_tables(ddl_con):
     assert isinstance(temp_tables, list)
     assert temp_table_name in temp_tables
     assert temp_table_name not in ddl_con.ddl.list_views()
-    assert temp_table_name not in ddl_con.ddl.list_temp_views()
     assert temp_table_name not in ddl_con.ddl.list_tables()
+    try:
+        assert temp_table_name not in ddl_con.ddl.list_temp_views()
+    except NotImplementedError:  # not all backends have list_temp_views
+        return
 
 
+@pytest.mark.never("mysql", reason="mysql does not support temporary views")
 def test_list_temp_views(ddl_con):
     # TODO: replace raw_sql with create_temp