get_current_catalog

Author: fresioAS

sqlmesh/core/engine_adapter/fabric.py

@@ -108,6 +108,10 @@ def _drop_catalog(self, catalog_name: exp.Identifier) -> None:
             # will fail with an 'Authentication Failed' error unless we close all connections here, which also clears all the threadlocal data
             self.close()
 
+    def get_current_catalog(self) -> t.Optional[str]:
+        """Return the adapter-managed catalog for Fabric's stateless sessions."""
+        return self._target_catalog or self._extra_config.get("database")
+
     def set_current_catalog(self, catalog_name: str) -> None:
         """
         Set the current catalog for Microsoft Fabric connections.

tests/core/engine_adapter/test_fabric.py

@@ -19,6 +19,50 @@ def adapter(make_mocked_engine_adapter: t.Callable) -> FabricEngineAdapter:
     return make_mocked_engine_adapter(FabricEngineAdapter)
 
 
+def test_get_current_catalog_uses_target_catalog_or_configured_database(
+    make_mocked_engine_adapter: t.Callable,
+):
+    adapter = make_mocked_engine_adapter(
+        FabricEngineAdapter,
+        database="default_catalog",
+    )
+
+    assert adapter.get_current_catalog() == "default_catalog"
+
+    adapter._target_catalog = "switched_catalog"
+
+    assert adapter.get_current_catalog() == "switched_catalog"
+
+    adapter._connection_pool.close()
+
+    assert adapter._connection_pool.get_attribute("target_catalog") is None
+    assert adapter.get_current_catalog() == "default_catalog"
+    adapter.cursor.execute.assert_not_called()
+
+
+def test_get_current_catalog_returns_none_without_target_or_database(
+    make_mocked_engine_adapter: t.Callable,
+):
+    adapter = make_mocked_engine_adapter(FabricEngineAdapter)
+
+    assert adapter.get_current_catalog() is None
+    adapter.cursor.execute.assert_not_called()
+
+
+def test_set_current_catalog_does_not_query_database(
+    make_mocked_engine_adapter: t.Callable,
+):
+    adapter = make_mocked_engine_adapter(
+        FabricEngineAdapter,
+        database="default_catalog",
+    )
+
+    adapter.set_current_catalog("new_catalog")
+
+    assert adapter.get_current_catalog() == "new_catalog"
+    adapter.cursor.execute.assert_not_called()
+
+
 def test_columns(adapter: FabricEngineAdapter):
     adapter.cursor.fetchall.return_value = [
         ("decimal_ps", "decimal", None, 5, 4),