address comments

Author: themisvaltinos

sqlmesh/core/config/scheduler.py

@@ -46,14 +46,6 @@ def create_state_sync(self, context: GenericContext) -> StateSync:
             The StateSync instance.
         """
 
-    @abc.abstractmethod
-    def get_default_catalog(self, context: GenericContext) -> t.Optional[str]:
-        """Returns the default catalog for the Scheduler.
-
-        Args:
-            context: The SQLMesh Context.
-        """
-
     @abc.abstractmethod
     def get_default_catalog_per_gateway(self, context: GenericContext) -> t.Dict[str, str]:
         """Returns the default catalog for each gateway.
@@ -74,7 +66,7 @@ def state_sync_fingerprint(self, context: GenericContext) -> str:
 class _EngineAdapterStateSyncSchedulerConfig(SchedulerConfig):
     def create_state_sync(self, context: GenericContext) -> StateSync:
         state_connection = (
-            context.config.get_state_connection(context.gateway) or context._connection_config
+            context.config.get_state_connection(context.gateway) or context.connection_config
         )
 
         warehouse_connection = context.config.get_connection(context.gateway)
@@ -118,7 +110,7 @@ def create_state_sync(self, context: GenericContext) -> StateSync:
 
     def state_sync_fingerprint(self, context: GenericContext) -> str:
         state_connection = (
-            context.config.get_state_connection(context.gateway) or context._connection_config
+            context.config.get_state_connection(context.gateway) or context.connection_config
         )
         return md5(
             [
@@ -140,19 +132,16 @@ def create_plan_evaluator(self, context: GenericContext) -> PlanEvaluator:
             state_sync=context.state_sync,
             snapshot_evaluator=context.snapshot_evaluator,
             create_scheduler=context.create_scheduler,
-            default_catalog=self.get_default_catalog(context),
+            default_catalog=context.default_catalog,
             console=context.console,
         )
 
-    def get_default_catalog(self, context: GenericContext) -> t.Optional[str]:
-        return context.engine_adapter.default_catalog
-
     def get_default_catalog_per_gateway(self, context: GenericContext) -> t.Dict[str, str]:
-        return {
-            name: adapter.default_catalog
-            for name, adapter in context.engine_adapters.items()
-            if adapter.default_catalog
-        }
+        default_catalogs_per_gateway: t.Dict[str, str] = {}
+        for gateway, adapter in context.engine_adapters.items():
+            if catalog := adapter.default_catalog:
+                default_catalogs_per_gateway[gateway] = catalog
+        return default_catalogs_per_gateway
 
 
 SCHEDULER_CONFIG_TO_TYPE = {

sqlmesh/core/context.py

@@ -369,7 +369,10 @@ def __init__(
         self._environment_statements: t.List[EnvironmentStatements] = []
         self._excluded_requirements: t.Set[str] = set()
         self._default_catalog: t.Optional[str] = None
+        self._default_catalog_per_gateway: t.Optional[t.Dict[str, str]] = None
         self._engine_adapter: t.Optional[EngineAdapter] = None
+        self._connection_config: t.Optional[ConnectionConfig] = None
+        self._test_connection_config: t.Optional[ConnectionConfig] = None
         self._linters: t.Dict[str, Linter] = {}
         self._loaded: bool = False
 
@@ -412,7 +415,7 @@ def __init__(
 
         self._concurrent_tasks = concurrent_tasks
         self._state_connection_config = (
-            self.config.get_state_connection(self.gateway) or self._connection_config
+            self.config.get_state_connection(self.gateway) or self.connection_config
         )
 
         self._snapshot_evaluator: t.Optional[SnapshotEvaluator] = None
@@ -440,7 +443,7 @@ def default_dialect(self) -> t.Optional[str]:
     def engine_adapter(self) -> EngineAdapter:
         """Returns the default engine adapter."""
         if self._engine_adapter is None:
-            self._engine_adapter = self._connection_config.create_engine_adapter()
+            self._engine_adapter = self.connection_config.create_engine_adapter()
         return self._engine_adapter
 
     @property
@@ -965,8 +968,8 @@ def requirements(self) -> t.Dict[str, str]:
 
     @property
     def default_catalog(self) -> t.Optional[str]:
-        if self._default_catalog is None:
-            self._default_catalog = self._scheduler.get_default_catalog(self)
+        if self._default_catalog is None and self.default_catalog_per_gateway:
+            self._default_catalog = self.default_catalog_per_gateway[self.selected_gateway]
         return self._default_catalog
 
     @python_api_analytics
@@ -1978,7 +1981,7 @@ def create_test(
 
         try:
             model_to_test = self.get_model(model, raise_if_missing=True)
-            test_adapter = self._test_connection_config.create_engine_adapter(
+            test_adapter = self.test_connection_config.create_engine_adapter(
                 register_comments_override=False
             )
 
@@ -2463,7 +2466,7 @@ def _run_plan_tests(
                 self.console.log_test_results(
                     result,
                     test_output,
-                    self._test_connection_config._engine_adapter.DIALECT,
+                    self.test_connection_config._engine_adapter.DIALECT,
                 )
             if not result.wasSuccessful():
                 raise PlanError(
@@ -2507,30 +2510,45 @@ def engine_adapters(self) -> t.Dict[str, EngineAdapter]:
     @cached_property
     def default_catalog_per_gateway(self) -> t.Dict[str, str]:
         """Returns the default catalogs for each engine adapter."""
-        if self.gateway_managed_virtual_layer:
-            return self._scheduler.get_default_catalog_per_gateway(self)
-        return {}
+        if self._default_catalog_per_gateway is None:
+            self._default_catalog_per_gateway = self._scheduler.get_default_catalog_per_gateway(
+                self
+            )
+        return self._default_catalog_per_gateway
 
     @cached_property
     def concurrent_tasks(self) -> int:
         if self._concurrent_tasks is None:
-            self._concurrent_tasks = self._connection_config.concurrent_tasks
+            self._concurrent_tasks = self.connection_config.concurrent_tasks
         return self._concurrent_tasks
 
     @cached_property
-    def _connection_config(self) -> ConnectionConfig:
-        return self.config.get_connection(self.gateway)
+    def connection_config(self) -> ConnectionConfig:
+        if self._connection_config is None:
+            self._connection_config = self.config.get_connection(self.selected_gateway)
+        return self._connection_config
 
     @cached_property
-    def _test_connection_config(self) -> ConnectionConfig:
-        return self.config.get_test_connection(
-            self.gateway, self.default_catalog, default_catalog_dialect=self.engine_adapter.DIALECT
-        )
+    def test_connection_config(self) -> ConnectionConfig:
+        if self._test_connection_config is None:
+            self._test_connection_config = self.config.get_test_connection(
+                self.gateway,
+                self.default_catalog,
+                default_catalog_dialect=self.engine_adapter.DIALECT,
+            )
+        return self._test_connection_config
 
     @cached_property
     def environment_catalog_mapping(self) -> RegexKeyDict:
+        engine_adapter = None
+        try:
+            engine_adapter = self.engine_adapter
+        except Exception:
+            pass
+
         if (
             self.config.environment_catalog_mapping
+            and engine_adapter
             and not self.engine_adapter.catalog_support.is_multi_catalog_supported
         ):
             raise SQLMeshError(

sqlmesh/integrations/github/cicd/controller.py

@@ -726,7 +726,7 @@ def conclusion_handler(
                 self._console.log_test_results(
                     result,
                     output,
-                    self._context._test_connection_config._engine_adapter.DIALECT,
+                    self._context.test_connection_config._engine_adapter.DIALECT,
                 )
                 test_summary = self._console.consume_captured_output()
                 test_title = "Tests Passed" if result.wasSuccessful() else "Tests Failed"

tests/core/test_config.py

@@ -742,7 +742,7 @@ def test_multi_gateway_config(tmp_path, mocker: MockerFixture):
 
     ctx = Context(paths=tmp_path, config=config)
 
-    assert isinstance(ctx._connection_config, RedshiftConnectionConfig)
+    assert isinstance(ctx.connection_config, RedshiftConnectionConfig)
     assert len(ctx.engine_adapters) == 3
     assert isinstance(ctx.engine_adapters["athena"], AthenaEngineAdapter)
     assert isinstance(ctx.engine_adapters["redshift"], RedshiftEngineAdapter)
@@ -782,7 +782,7 @@ def test_multi_gateway_single_threaded_config(tmp_path):
     )
 
     ctx = Context(paths=tmp_path, config=config)
-    assert isinstance(ctx._connection_config, DuckDBConnectionConfig)
+    assert isinstance(ctx.connection_config, DuckDBConnectionConfig)
     assert len(ctx.engine_adapters) == 2
     assert ctx.engine_adapter == ctx._get_engine_adapter("duckdb")
     assert isinstance(ctx.engine_adapters["athena"], AthenaEngineAdapter)

tests/core/test_integration.py

@@ -4872,13 +4872,11 @@ def test_multi(mocker):
     assert context.fetchdf("select * from after_1").to_dict()["repo_1"][0] == "repo_1"
     assert context.fetchdf("select * from after_2").to_dict()["repo_2"][0] == "repo_2"
 
-    adapter = context.engine_adapter
     context = Context(
         paths=["examples/multi/repo_1"],
         state_sync=context.state_sync,
         gateway="memory",
     )
-    context._engine_adapter = adapter
 
     model = context.get_model("bronze.a")
     assert model.project == "repo_1"
@@ -4935,6 +4933,8 @@ def test_multi_virtual_layer(copy_to_temp_path):
     )
 
     context = Context(paths=paths, config=config)
+    assert context.default_catalog_per_gateway == {"first": "db_1", "second": "db_2"}
+    assert len(context.engine_adapters) == 2
 
     # For the model without gateway the default should be used and the gateway variable should overide the global
     assert (
@@ -5064,39 +5064,6 @@ def test_multi_virtual_layer(copy_to_temp_path):
         context.apply(plan)
 
 
-def test_multi_virtual_layer_catalogs(copy_to_temp_path):
-    paths = copy_to_temp_path("tests/fixtures/multi_virtual_layer")
-    path = Path(paths[0])
-    first_db_path = str(path / "db_1.db")
-    second_db_path = str(path / "db_2.db")
-
-    config = Config(
-        gateways={
-            "first": GatewayConfig(
-                connection=DuckDBConnectionConfig(database=first_db_path),
-                variables={"overriden_var": "gateway_1"},
-            ),
-            "second": GatewayConfig(
-                connection=DuckDBConnectionConfig(database=second_db_path),
-                variables={"overriden_var": "gateway_2"},
-            ),
-        },
-        model_defaults=ModelDefaultsConfig(dialect="duckdb"),
-        model_naming=NameInferenceConfig(infer_names=True),
-        default_gateway="first",
-        variables={"overriden_var": "global", "global_one": 88},
-    )
-
-    # With gateway_managed_virtual_layer to False the catalogs won't be retrieved
-    context = Context(paths=paths, config=config)
-    assert context.default_catalog_per_gateway == {}
-
-    config.gateway_managed_virtual_layer = True
-    context = Context(paths=paths, config=config)
-    assert context.default_catalog_per_gateway == {"first": "db_1", "second": "db_2"}
-    assert len(context.engine_adapters) == 2
-
-
 def test_multi_dbt(mocker):
     context = Context(paths=["examples/multi_dbt/bronze", "examples/multi_dbt/silver"])
     context._new_state_sync().reset(default_catalog=context.default_catalog)

tests/core/test_test.py

@@ -54,7 +54,7 @@ def _create_test(
         test_name=test_name,
         model=model,
         models=context._models,
-        engine_adapter=context._test_connection_config.create_engine_adapter(
+        engine_adapter=context.test_connection_config.create_engine_adapter(
             register_comments_override=False
         ),
         dialect=context.config.dialect,

web/server/api/endpoints/commands.py

@@ -159,7 +159,7 @@ async def test(
     context.console.log_test_results(
         result,
         test_output.getvalue(),
-        context._test_connection_config._engine_adapter.DIALECT,
+        context.test_connection_config._engine_adapter.DIALECT,
     )
 
     def _test_path(test: ModelTest) -> t.Optional[str]: