Enable dlt asset metadata via translator

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt/dlt/asset_decorator.py

@@ -99,6 +99,7 @@ def github_reactions_dagster_assets(context: AssetExecutionContext, dlt: DltDags
                     META_KEY_SOURCE: dlt_source,
                     META_KEY_PIPELINE: dlt_pipeline,
                     META_KEY_TRANSLATOR: dlt_dagster_translator,
+                    **dlt_dagster_translator.get_metadata(dlt_source_resource),
                 },
             )
             for dlt_source_resource in dlt_source.selected_resources.values()

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt/dlt/translator.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Iterable, Optional
+from typing import Any, Iterable, Mapping, Optional
 
 from dagster import (
     AssetKey,
@@ -51,3 +51,15 @@ def get_auto_materialize_policy(self, resource: DltResource) -> Optional[AutoMat
 
         """
         return None
+
+    @public
+    def get_metadata(self, resource: DltResource) -> Mapping[str, Any]:
+        """Defines resource specific metadata.
+
+        Args:
+            resource (DltResource): dlt resource / transformer
+
+        Returns:
+            Mapping[str, Any]: The custom metadata entries for this resource.
+        """
+        return {}

python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt_tests/dlt_tests/test_dlt.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Any, Mapping, Optional
 
 import duckdb
 from dagster import (
@@ -205,3 +205,46 @@ def example_pipeline_assets(
     assert not res.success
     asset_materizations = res.get_asset_materialization_events()
     assert len(asset_materizations) == 0
+
+
+def test_asset_metadata(dlt_pipeline: Pipeline) -> None:
+    class CustomDagsterDltTranslator(DagsterDltTranslator):
+        metadata_by_resource_name = {
+            "repos": {"mode": "upsert", "primary_key": "id"},
+            "repo_issues": {"mode": "upsert", "primary_key": ["repo_id", "issue_id"]},
+        }
+
+        def get_metadata(self, resource: DltResource) -> Mapping[str, Any]:
+            return self.metadata_by_resource_name.get(resource.name, {})
+
+    @dlt_assets(
+        dlt_source=pipeline(),
+        dlt_pipeline=dlt_pipeline,
+        dlt_dagster_translator=CustomDagsterDltTranslator(),
+    )
+    def example_pipeline_assets(
+        context: AssetExecutionContext, dlt_pipeline_resource: DagsterDltResource
+    ):
+        yield from dlt_pipeline_resource.run(context=context)
+
+    first_asset_metadata = next(iter(example_pipeline_assets.metadata_by_key.values()))
+    dagster_dlt_source = first_asset_metadata.get("dagster_dlt/source")
+    dagster_dlt_pipeline = first_asset_metadata.get("dagster_dlt/pipeline")
+    dagster_dlt_translator = first_asset_metadata.get("dagster_dlt/translator")
+
+    assert example_pipeline_assets.metadata_by_key == {
+        AssetKey("dlt_pipeline_repos"): {
+            "dagster_dlt/source": dagster_dlt_source,
+            "dagster_dlt/pipeline": dagster_dlt_pipeline,
+            "dagster_dlt/translator": dagster_dlt_translator,
+            "mode": "upsert",
+            "primary_key": "id",
+        },
+        AssetKey("dlt_pipeline_repo_issues"): {
+            "dagster_dlt/source": dagster_dlt_source,
+            "dagster_dlt/pipeline": dagster_dlt_pipeline,
+            "dagster_dlt/translator": dagster_dlt_translator,
+            "mode": "upsert",
+            "primary_key": ["repo_id", "issue_id"],
+        },
+    }