chore: merge source for PR

Author: lucargir

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.16.0 (2025-04-30)
+
+* feat: update sqlmesh pin to 0.178.2 (#45) ([27245d9](https://github.com/opensource-observer/oso/commit/27245d9)), closes [#45](https://github.com/opensource-observer/oso/issues/45)
+
 ## 0.15.0 (2025-04-23)
 
 * feat: allow passing in custom context (#42) ([63bd837](https://github.com/opensource-observer/oso/commit/63bd837)), closes [#42](https://github.com/opensource-observer/oso/issues/42)

dagster_sqlmesh/conftest.py

@@ -6,15 +6,12 @@
 import typing as t
 
 import pytest
-from sqlmesh.core.config import (
-    Config as SQLMeshConfig,
-    DuckDBConnectionConfig,
-    GatewayConfig,
-    ModelDefaultsConfig,
-)
 
 from dagster_sqlmesh.config import SQLMeshContextConfig
-from dagster_sqlmesh.testing import SQLMeshTestContext
+from dagster_sqlmesh.testing import (
+    SQLMeshTestContext,
+    setup_testing_sqlmesh_context_config,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -41,23 +38,19 @@ def sample_sqlmesh_project() -> t.Iterator[str]:
         # Initialize the "source" data
         yield str(project_dir)
 
+@pytest.fixture
+def sample_sqlmesh_db_path(sample_sqlmesh_project: str) -> t.Iterator[str]:
+    db_path = os.path.join(sample_sqlmesh_project, "db.db")
+    yield db_path
+
+@pytest.fixture
+def sample_sqlmesh_test_context_config(sample_sqlmesh_project: str, sample_sqlmesh_db_path: str) -> t.Iterator[SQLMeshContextConfig]:
+    yield setup_testing_sqlmesh_context_config(db_path=sample_sqlmesh_db_path, project_path=sample_sqlmesh_project)
 
 @pytest.fixture
 def sample_sqlmesh_test_context(
-    sample_sqlmesh_project: str,
+    sample_sqlmesh_project: str, sample_sqlmesh_test_context_config: SQLMeshContextConfig, sample_sqlmesh_db_path: str
 ) -> t.Iterator[SQLMeshTestContext]:
-    db_path = os.path.join(sample_sqlmesh_project, "db.db")
-    config = SQLMeshConfig(
-        gateways={
-            "local": GatewayConfig(connection=DuckDBConnectionConfig(database=db_path)),
-        },
-        default_gateway="local",
-        model_defaults=ModelDefaultsConfig(dialect="duckdb"),
-    )
-    config_as_dict = config.dict()
-    context_config = SQLMeshContextConfig(
-        path=sample_sqlmesh_project, gateway="local", config_override=config_as_dict
-    )
-    test_context = SQLMeshTestContext(db_path=db_path, context_config=context_config)
+    test_context = SQLMeshTestContext(db_path=sample_sqlmesh_db_path, context_config=sample_sqlmesh_test_context_config)
     test_context.initialize_test_source()
     yield test_context

dagster_sqlmesh/console.py

@@ -405,14 +405,14 @@ def publish_known_event(self, event_name: str, **kwargs: t.Any) -> None:
 
     def publish(self, event: ConsoleEvent) -> None:
         self.logger.debug(
-            f"EventConsole[{self.id}]: sending event to {len(self._handlers)}"
+            f"EventConsole[{self.id}]: sending event {event.__class__.__name__} to {len(self._handlers)}"
         )
         for handler in self._handlers.values():
             handler(event)
 
     def publish_unknown_event(self, event_name: str, **kwargs: t.Any) -> None:
         self.logger.debug(
-            f"EventConsole[{self.id}]: sending unknown event to {len(self._handlers)}"
+            f"EventConsole[{self.id}]: sending unknown '{event_name}' event to {len(self._handlers)} handlers"
         )
         self.logger.debug(f"EventConsole[{self.id}]: unknown event {event_name} {kwargs}")
 

dagster_sqlmesh/controller/base.py

@@ -179,7 +179,17 @@ def run_sqlmesh_thread(
             default_catalog: str,
         ) -> None:
             logger.debug("dagster-sqlmesh: thread started")
+
+            def auto_execute_plan(event: ConsoleEvent):
+                if isinstance(event, Plan):
+                    try:
+                        event.plan_builder.apply()
+                    except Exception as e:
+                        controller.console.exception(e)
+                return None
+
             try:
+                controller.console.add_handler(auto_execute_plan)
                 builder = t.cast(
                     PlanBuilder,
                     context.plan_builder(
@@ -192,7 +202,7 @@ def run_sqlmesh_thread(
                     builder,
                     auto_apply=True,
                     default_catalog=default_catalog,
-                )
+                ) 
             except Exception as e:
                 controller.console.exception(e)
             except:  # noqa: E722
@@ -219,16 +229,18 @@ def run_sqlmesh_thread(
             thread.start()
 
             self.logger.debug("waiting for events")
-            for event in generator.events(thread):
-                match event:
-                    case ConsoleException(exception=e):
-                        raise e
-                    case Plan(plan_builder=plan_builder, auto_apply=auto_apply):
-                        if auto_apply:
-                            plan_builder.apply()
-                        yield event
-                    case _:
-                        yield event
+            try:
+                for event in generator.events(thread):
+                    match event:
+                        case ConsoleException(exception=e):
+                            raise e
+                        case _:
+                            yield event
+            except Exception as e:
+                import traceback
+                print("An exception occurred:")
+                print(traceback.format_exc())
+                raise
 
             thread.join()