fix: use custom translator properly in sample dagster

Author: ravenac95

sample/dagster_project/definitions.py

@@ -8,6 +8,7 @@
     AssetKey,
     Definitions,
     MaterializeResult,
+    ResourceParam,
     asset,
     define_asset_job,
 )
@@ -27,12 +28,18 @@
 SQLMESH_CACHE_PATH = os.path.join(SQLMESH_PROJECT_PATH, ".cache")
 DUCKDB_PATH = os.path.join(CURR_DIR, "../../db.db")
 
-sqlmesh_config = SQLMeshContextConfig(
+class CustomSQLMeshContextConfig(SQLMeshContextConfig):
+    custom_key: str
+
+    def get_translator(self):
+        return RewrittenSQLMeshTranslator(self.custom_key)
+
+sqlmesh_config = CustomSQLMeshContextConfig(
     path=SQLMESH_PROJECT_PATH, 
     gateway="local",
+    custom_key="custom_sqlmesh_prefix"
 )
 
-
 class RewrittenSQLMeshTranslator(SQLMeshDagsterTranslator):
     """A contrived SQLMeshDagsterTranslator that flattens the catalog of the
     sqlmesh project and only uses the table db and name
@@ -85,7 +92,7 @@ def test_source() -> pl.DataFrame:
     )
 
 
-@asset(deps=[AssetKey(["sqlmesh", "full_model"])])
+@asset(deps=[AssetKey(["custom_sqlmesh_prefix", "full_model"])])
 def post_full_model() -> pl.DataFrame:
     """An asset that depends on the `full_model` asset from the sqlmesh project.
     This is used to test that the sqlmesh assets are correctly materialized and
@@ -109,7 +116,7 @@ def post_full_model() -> pl.DataFrame:
     enabled_subsetting=True,
 )
 def sqlmesh_project(
-    context: AssetExecutionContext, sqlmesh: SQLMeshResource, sqlmesh_config: SQLMeshContextConfig
+    context: AssetExecutionContext, sqlmesh: SQLMeshResource, sqlmesh_config: ResourceParam[SQLMeshContextConfig]
 ) -> t.Iterator[MaterializeResult[t.Any]]:
     yield from sqlmesh.run(context, config=sqlmesh_config)