Fixed broken usage of instance

Author: ravenac95

dagster_sqlmesh/controller/base.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from math import log
 import typing as t
 import logging
 import threading
@@ -350,13 +351,19 @@ def setup_with_config(
         controller = cls(
             console=console,
             config=config,
+            log_override=log_override,
         )
         return controller
 
-    def __init__(self, config: SQLMeshContextConfig, console: EventConsole):
+    def __init__(
+        self,
+        config: SQLMeshContextConfig,
+        console: EventConsole,
+        log_override: t.Optional[logging.Logger] = None,
+    ):
         self.config = config
         self.console = console
-        self.logger = logger
+        self.logger = log_override or logger
         self._context_open = False
 
     def set_logger(self, logger: logging.Logger):
@@ -379,7 +386,10 @@ def _create_context(self):
         return Context(**options)
 
     @contextmanager
-    def instance(self, environment: str):
+    def instance(self, environment: str, component: str = "unknown"):
+        self.logger.info(
+            f"Opening sqlmesh instance for env={environment} component={component}"
+        )
         if self._context_open:
             raise Exception("Only one sqlmesh instance at a time")
 
@@ -390,6 +400,9 @@ def instance(self, environment: str):
                 environment, self.console, self.config, context, self.logger
             )
         finally:
+            self.logger.info(
+                f"Closing sqlmesh instance for env={environment} component={component}"
+            )
             self._context_open = False
             context.close()
 
@@ -398,7 +411,7 @@ def run(
         environment: str,
         **run_options: t.Unpack[RunOptions],
     ):
-        with self.instance(environment) as mesh:
+        with self.instance(environment, "run") as mesh:
             yield from mesh.run(**run_options)
 
     def plan(
@@ -408,7 +421,7 @@ def plan(
         default_catalog: t.Optional[str],
         plan_options: PlanOptions,
     ):
-        with self.instance(environment) as mesh:
+        with self.instance(environment, "plan") as mesh:
             yield from mesh.plan(categorizer, default_catalog, **plan_options)
 
     def plan_and_run(
@@ -419,7 +432,7 @@ def plan_and_run(
         plan_options: t.Optional[PlanOptions] = None,
         run_options: t.Optional[RunOptions] = None,
     ):
-        with self.instance(environment) as mesh:
+        with self.instance(environment, "plan_and_run") as mesh:
             yield from mesh.plan_and_run(
                 categorizer=categorizer,
                 default_catalog=default_catalog,

dagster_sqlmesh/controller/dagster.py

@@ -22,7 +22,7 @@ class DagsterSQLMeshController(SQLMeshController):
     def to_asset_outs(
         self, environment: str, translator: SQLMeshDagsterTranslator
     ) -> SQLMeshMultiAssetOptions:
-        with self.instance(environment) as instance:
+        with self.instance(environment, "to_asset_outs") as instance:
             context = instance.context
             dag = context.dag
             output = SQLMeshMultiAssetOptions()

dagster_sqlmesh/resource.py

@@ -248,8 +248,7 @@ def run(
                 context, models_map, dag, "sqlmesh: "
             )
 
-            for event in controller.plan_and_run(
-                environment=environment,
+            for event in mesh.plan_and_run(
                 plan_options=plan_options,
                 run_options=run_options,
             ):

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dagster-sqlmesh"
-version = "0.3.1-dev1"
+version = "0.3.1"
 description = ""
 authors = ["Reuven Gonzales <reuven@karibalabs.co>"]
 license = "Apache 2.0"