feat(DB engine spec): get_catalog_names

superset/db_engine_specs/base.py

@@ -374,6 +374,14 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
     # a custom `adjust_engine_params` method.
     supports_dynamic_schema = False
 
+    # Does the DB support catalogs? A catalog here is a group of schemas, and has
+    # different names depending on the DB: BigQuery calles it a "project", Postgres calls
+    # it a "database", Trino calls it a "catalog", etc.
+    supports_catalog = False
+
+    # Can the catalog be changed on a per-query basis?
+    supports_dynamic_catalog = False
+
     @classmethod
     def supports_url(cls, url: URL) -> bool:
         """
@@ -1091,6 +1099,20 @@ def patch(cls) -> None:
         TODO: Improve docstring and refactor implementation in Hive
         """
 
+    @classmethod
+    def get_catalog_names(  # pylint: disable=unused-argument
+        cls,
+        database: Database,
+        inspector: Inspector,
+    ) -> List[str]:
+        """
+        Get all catalogs from database.
+
+        This needs to be implemented per database, since SQLAlchemy doesn't offer an
+        abstraction.
+        """
+        return []
+
     @classmethod
     def get_schema_names(cls, inspector: Inspector) -> List[str]:
         """

superset/db_engine_specs/postgres.py

@@ -23,6 +23,7 @@
 from flask_babel import gettext as __
 from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION, ENUM, JSON
 from sqlalchemy.dialects.postgresql.base import PGInspector
+from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import URL
 from sqlalchemy.types import Date, DateTime, String
 
@@ -291,6 +292,27 @@ def query_cost_formatter(
     ) -> List[Dict[str, str]]:
         return [{k: str(v) for k, v in row.items()} for row in raw_cost]
 
+    @classmethod
+    def get_catalog_names(
+        cls,
+        database: "Database",
+        inspector: Inspector,
+    ) -> List[str]:
+        """
+        Return all catalogs.
+
+        In Postgres, a catalog is called a "database".
+        """
+        return sorted(
+            catalog
+            for (catalog,) in inspector.bind.execute(
+                """
+SELECT datname FROM pg_database
+WHERE datistemplate = false;
+            """
+            ).fetchall()
+        )
+
     @classmethod
     def get_table_names(
         cls, database: "Database", inspector: PGInspector, schema: Optional[str]

tests/integration_tests/db_engine_specs/postgres_tests.py

@@ -17,13 +17,15 @@
 from textwrap import dedent
 from unittest import mock
 
+from flask.ctx import AppContext
 from sqlalchemy import column, literal_column
 from sqlalchemy.dialects import postgresql
 
 from superset.db_engine_specs import load_engine_specs
 from superset.db_engine_specs.postgres import PostgresEngineSpec
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.models.sql_lab import Query
+from superset.utils.database import get_example_database
 from tests.integration_tests.db_engine_specs.base_tests import TestDbEngineSpec
 from tests.integration_tests.fixtures.certificates import ssl_certificate
 from tests.integration_tests.fixtures.database import default_db_extra
@@ -514,3 +516,19 @@ def test_base_parameters_mixin():
         },
         "required": ["database", "host", "port", "username"],
     }
+
+
+def test_get_catalog_names(app_context: AppContext) -> None:
+    """
+    Test the ``get_catalog_names`` method.
+    """
+    database = get_example_database()
+
+    if database.backend != "postgresql":
+        return
+
+    with database.get_inspector_with_context() as inspector:
+        assert PostgresEngineSpec.get_catalog_names(database, inspector) == [
+            "postgres",
+            "superset",
+        ]