Skip to content

Commit 5a84331

Browse files
committed
fix: optional translator, define otherwise in config and use sqlglot in translator
1 parent 514446f commit 5a84331

File tree

3 files changed

+12
-8
lines changed

3 files changed

+12
-8
lines changed

dagster_sqlmesh/controller/base.py

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -427,7 +427,7 @@ def setup(
427427
context_factory: ContextFactory[ContextCls],
428428
gateway: str = "local",
429429
log_override: logging.Logger | None = None,
430-
translator: SQLMeshDagsterTranslator = SQLMeshDagsterTranslator(),
430+
translator: SQLMeshDagsterTranslator | None = None,
431431
) -> t.Self:
432432
return cls.setup_with_config(
433433
config=SQLMeshContextConfig(path=path, gateway=gateway),
@@ -443,9 +443,11 @@ def setup_with_config(
443443
config: SQLMeshContextConfig,
444444
context_factory: ContextFactory[ContextCls] = DEFAULT_CONTEXT_FACTORY,
445445
log_override: logging.Logger | None = None,
446-
translator: SQLMeshDagsterTranslator = SQLMeshDagsterTranslator(),
446+
translator: SQLMeshDagsterTranslator | None = None,
447447
) -> t.Self:
448448
console = EventConsole(log_override=log_override) # type: ignore
449+
if not translator:
450+
translator = SQLMeshDagsterTranslator()
449451
controller = cls(
450452
console=console,
451453
config=config,
@@ -460,8 +462,8 @@ def __init__(
460462
config: SQLMeshContextConfig,
461463
console: EventConsole,
462464
context_factory: ContextFactory[ContextCls],
465+
translator: SQLMeshDagsterTranslator,
463466
log_override: logging.Logger | None = None,
464-
translator: SQLMeshDagsterTranslator = SQLMeshDagsterTranslator(),
465467
) -> None:
466468
self.config = config
467469
self.console = console

dagster_sqlmesh/translator.py

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
1-
import re
21
from collections.abc import Sequence
32

43
from dagster import AssetKey
4+
from sqlglot import exp
55
from sqlmesh.core.context import Context
66
from sqlmesh.core.model import Model
77

@@ -15,8 +15,10 @@ def get_asset_key(self, context: Context, fqn: str) -> AssetKey:
1515
return AssetKey(path)
1616

1717
def get_asset_key_name(self, fqn: str) -> Sequence[str]:
18-
asset_path = re.findall(r"[A-Za-z0-9_\-]+", fqn)
19-
return asset_path
18+
table = exp.to_table(fqn)
19+
asset_key_name = [table.catalog, table.db, table.name]
20+
21+
return asset_key_name
2022

2123
def get_asset_key_str(self, fqn: str) -> str:
2224
# This is an internal identifier used to map outputs and dependencies

sample/dagster_project/definitions.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
sqlmesh_config = SQLMeshContextConfig(path=SQLMESH_PROJECT_PATH, gateway="local")
2222

2323

24-
@asset
24+
@asset(key=["db", "sources", "reset_asset"])
2525
def reset_asset() -> MaterializeResult:
2626
"""An asset used for testing this entire workflow. If the duckdb database is
2727
found, this will delete it. This allows us to continously test this dag if
@@ -34,7 +34,7 @@ def reset_asset() -> MaterializeResult:
3434
return MaterializeResult(metadata={"deleted": deleted})
3535

3636

37-
@asset(deps=[reset_asset])
37+
@asset(deps=[reset_asset], key=["db", "sources", "test_source"])
3838
def test_source() -> pl.DataFrame:
3939
"""Sets up the `test_source` table in duckdb that one of the sample sqlmesh
4040
models depends on"""

0 commit comments

Comments
 (0)