fix: rename translator and restore test resource naming

Author: lucargir

dagster_sqlmesh/asset.py

@@ -26,11 +26,11 @@ def sqlmesh_to_multi_asset_options(
     controller = DagsterSQLMeshController.setup_with_config(
         config=config, context_factory=context_factory
     )
-    dagster_sqlmesh_translator = config.get_translator()
+    translator = config.get_translator()
 
     conversion = controller.to_asset_outs(
         environment,
-        dagster_sqlmesh_translator=dagster_sqlmesh_translator,
+        translator=translator,
     )
     return conversion
 

dagster_sqlmesh/config.py

@@ -1,22 +1,22 @@
 import inspect
+import typing as t
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
 
 from dagster import Config
 from pydantic import Field
 from sqlmesh.core.config import Config as MeshConfig
 from sqlmesh.core.config.loader import load_configs
 
-if TYPE_CHECKING:
+if t.TYPE_CHECKING:
     from dagster_sqlmesh.translator import SQLMeshDagsterTranslator
 
 
 @dataclass
 class ConfigOverride:
-    config_as_dict: dict[str, Any]
+    config_as_dict: dict[str, t.Any]
 
-    def dict(self) -> dict[str, Any]:
+    def dict(self) -> dict[str, t.Any]:
         return self.config_as_dict
 
 
@@ -34,7 +34,7 @@ class SQLMeshContextConfig(Config):
 
     path: str
     gateway: str
-    config_override: dict[str, Any] | None = Field(default_factory=lambda: None)
+    config_override: dict[str, t.Any] | None = Field(default_factory=lambda: None)
     translator_class_name: str = Field(
         default="dagster_sqlmesh.translator.SQLMeshDagsterTranslator",
         description="Fully qualified class name of the SQLMesh Dagster translator to use"

dagster_sqlmesh/controller/dagster.py

@@ -22,7 +22,7 @@ class DagsterSQLMeshController(SQLMeshController[ContextCls]):
     def to_asset_outs(
         self,
         environment: str,
-        dagster_sqlmesh_translator: SQLMeshDagsterTranslator,
+        translator: SQLMeshDagsterTranslator,
     ) -> SQLMeshMultiAssetOptions:
         """Loads all the asset outs of the current sqlmesh environment. If a
         cache is provided, it will be tried first to load the asset outs."""
@@ -35,42 +35,42 @@ def to_asset_outs(
             context = instance.context
 
             for model, deps in instance.non_external_models_dag():
-                asset_key = dagster_sqlmesh_translator.get_asset_key(context=context, fqn=model.fqn)
+                asset_key = translator.get_asset_key(context=context, fqn=model.fqn)
                 asset_key_str = asset_key.to_user_string()
                 model_deps = [
                     SQLMeshModelDep(fqn=dep, model=context.get_model(dep))
                     for dep in deps
                 ]
                 internal_asset_deps: set[str] = set()
-                asset_tags = dagster_sqlmesh_translator.get_tags(context, model)
+                asset_tags = translator.get_tags(context, model)
 
                 for dep in model_deps:
                     if dep.model:
-                        dep_asset_key_str = dagster_sqlmesh_translator.get_asset_key(
+                        dep_asset_key_str = translator.get_asset_key(
                             context, dep.model.fqn
                         ).to_user_string()
 
                         internal_asset_deps.add(dep_asset_key_str)
                     else:
-                        table = dagster_sqlmesh_translator.get_asset_key_str(dep.fqn)
-                        key = dagster_sqlmesh_translator.get_asset_key(
+                        table = translator.get_asset_key_str(dep.fqn)
+                        key = translator.get_asset_key(
                             context, dep.fqn
                         ).to_user_string()
                         internal_asset_deps.add(key)
 
                         # create an external dep
-                        deps_map[table] = dagster_sqlmesh_translator.create_asset_dep(key=key)
+                        deps_map[table] = translator.create_asset_dep(key=key)
 
-                model_key = dagster_sqlmesh_translator.get_asset_key_str(model.fqn)
-                asset_outs[model_key] = dagster_sqlmesh_translator.create_asset_out(
+                model_key = translator.get_asset_key_str(model.fqn)
+                asset_outs[model_key] = translator.create_asset_out(
                     model_key=model_key,
                     asset_key=asset_key_str,
                     tags=asset_tags,
                     is_required=False,
-                    group_name=dagster_sqlmesh_translator.get_group_name(context, model),
+                    group_name=translator.get_group_name(context, model),
                     kinds={
                         "sqlmesh",
-                        dagster_sqlmesh_translator.get_context_dialect(context).lower(),
+                        translator.get_context_dialect(context).lower(),
                     },
                 )
                 internal_asset_deps_map[model_key] = internal_asset_deps

dagster_sqlmesh/resource.py

@@ -331,7 +331,7 @@ def __init__(
         models_map: dict[str, Model],
         dag: DAG[t.Any],
         prefix: str,
-        dagster_sqlmesh_translator: "SQLMeshDagsterTranslator",
+        translator: "SQLMeshDagsterTranslator",
         is_testing: bool = False,
         materializations_enabled: bool = True,
     ) -> None:
@@ -344,7 +344,7 @@ def __init__(
             models_map: A mapping of model names to their SQLMesh model instances.
             dag: The directed acyclic graph representing the SQLMesh models.
             prefix: A prefix to use for all asset keys generated by this handler.
-            dagster_sqlmesh_translator: The SQLMesh Dagster translator instance.
+            translator: The SQLMesh Dagster translator instance.
             is_testing: Whether the handler is being used in a testing context.
             materializations_enabled: Whether the handler is to generate
                 materializations, this should be disabled if you with to run a
@@ -355,7 +355,7 @@ def __init__(
         self._prefix = prefix
         self._context = context
         self._logger = context.log
-        self._translator = dagster_sqlmesh_translator
+        self._translator = translator
         self._tracker = MaterializationTracker(
             sorted_dag=dag.sorted[:], logger=self._logger
         )
@@ -698,7 +698,7 @@ def create_event_handler(
             dag=dag,
             models_map=models_map,
             prefix=prefix,
-            dagster_sqlmesh_translator=translator,
+            translator=translator,
             is_testing=is_testing,
             materializations_enabled=materializations_enabled,
         )

dagster_sqlmesh/test_asset.py

@@ -4,6 +4,6 @@
 def test_sqlmesh_context_to_asset_outs(sample_sqlmesh_test_context: SQLMeshTestContext):
     controller = sample_sqlmesh_test_context.create_controller()
     translator = sample_sqlmesh_test_context.context_config.get_translator()
-    outs = controller.to_asset_outs("dev", dagster_sqlmesh_translator=translator)
+    outs = controller.to_asset_outs("dev", translator=translator)
     assert len(list(outs.deps)) == 1
     assert len(outs.outs) == 10

dagster_sqlmesh/test_resource.py

@@ -10,15 +10,15 @@
     PlanOrRunFailedError,
 )
 from dagster_sqlmesh.testing import setup_testing_sqlmesh_test_context
-from dagster_sqlmesh.testing.context import SQLMeshTestContext, SQLMeshTestResource
+from dagster_sqlmesh.testing.context import SQLMeshTestContext, TestSQLMeshResource
 
 
 @dataclass(kw_only=True)
 class SQLMeshResourceInitialization:
     dagster_instance: dg.DagsterInstance
     dagster_context: dg.AssetExecutionContext
     test_context: SQLMeshTestContext
-    resource: SQLMeshTestResource
+    resource: TestSQLMeshResource
 
 
 def setup_sqlmesh_resource(