dict

python_modules/dagster/dagster/_core/definitions/metadata/__init__.py

@@ -1323,12 +1323,15 @@ def namespace(cls) -> str:
         return "dagster"
 
 
+DEFAULT_SOURCE_FILE_KEY = "asset_definition"
+
+
 class SourceDataMetadataSet(NamespacedMetadataSet):
     """Metadata entries that apply to asset definitions and which specify the source code location
     for the asset.
     """
 
-    source_paths: List[SourcePathMetadataSet] = []
+    source_paths: Dict[str, SourcePathMetadataSet] = {}
 
     @classmethod
     def namespace(cls) -> str:
@@ -1355,7 +1358,7 @@ def _with_code_source_single_definition(
     source_path = source_path_from_fn(base_fn)
 
     if source_path:
-        source_metadata = SourceDataMetadataSet(source_paths=[source_path])
+        source_metadata = SourceDataMetadataSet(source_paths={DEFAULT_SOURCE_FILE_KEY: source_path})
         for key in assets_def.keys:
             metadata_by_key[key] = {**metadata_by_key.get(key, {}), **source_metadata}
 

python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_defs_source_metadata.py

@@ -2,7 +2,11 @@
 from typing import cast
 
 from dagster import AssetsDefinition, load_assets_from_modules
-from dagster._core.definitions.metadata import SourcePathMetadataSet, with_code_source
+from dagster._core.definitions.metadata import (
+    DEFAULT_SOURCE_FILE_KEY,
+    SourcePathMetadataSet,
+    with_code_source,
+)
 from dagster._utils import file_relative_path
 
 
@@ -57,10 +61,12 @@ def test_asset_code_origins() -> None:
                 assert "dagster/source_paths" in asset.metadata_by_key[key]
                 assert len(asset.metadata_by_key[key]["dagster/source_paths"]) == 1
                 assert isinstance(
-                    asset.metadata_by_key[key]["dagster/source_paths"][0], SourcePathMetadataSet
+                    asset.metadata_by_key[key]["dagster/source_paths"][DEFAULT_SOURCE_FILE_KEY],
+                    SourcePathMetadataSet,
                 )
                 meta = cast(
-                    SourcePathMetadataSet, asset.metadata_by_key[key]["dagster/source_paths"][0]
+                    SourcePathMetadataSet,
+                    asset.metadata_by_key[key]["dagster/source_paths"][DEFAULT_SOURCE_FILE_KEY],
                 )
 
                 assert meta.path_to_module == expected_module_path

python_modules/dagster/dagster_tests/definitions_tests/metadata_tests/test_source_metadata_set.py

@@ -1,36 +1,38 @@
-import pytest
 from dagster._core.definitions.events import AssetMaterialization
-from dagster._core.definitions.metadata import SourceDataMetadataSet, SourcePathMetadataSet
-from dagster._core.test_utils import raise_exception_on_warnings
-
-
-@pytest.fixture(autouse=True)
-def error_on_warning() -> None:
-    raise_exception_on_warnings()
+from dagster._core.definitions.metadata import (
+    DEFAULT_SOURCE_FILE_KEY,
+    SourceDataMetadataSet,
+    SourcePathMetadataSet,
+)
 
 
 def test_source_metadata_set() -> None:
     source_metadata = SourceDataMetadataSet(
-        source_paths=[
-            SourcePathMetadataSet(
+        source_paths={
+            DEFAULT_SOURCE_FILE_KEY: SourcePathMetadataSet(
                 path_to_module="/Users/dagster/Documents/my_module",
                 path_from_module="assets/my_asset.py",
                 line_number=12,
             )
-        ]
+        }
     )
 
     dict_source_metadata = dict(source_metadata)
     assert dict_source_metadata == {"dagster/source_paths": source_metadata.source_paths}
     assert len(dict_source_metadata["dagster/source_paths"]) == 1
-    assert isinstance(dict_source_metadata["dagster/source_paths"][0], SourcePathMetadataSet)
+    assert isinstance(
+        dict_source_metadata["dagster/source_paths"][DEFAULT_SOURCE_FILE_KEY], SourcePathMetadataSet
+    )
     AssetMaterialization(asset_key="a", metadata=dict_source_metadata)
 
     splat_source_metadata = {**source_metadata}
     assert splat_source_metadata == {"dagster/source_paths": source_metadata.source_paths}
     assert len(splat_source_metadata["dagster/source_paths"]) == 1
-    assert isinstance(splat_source_metadata["dagster/source_paths"][0], SourcePathMetadataSet)
+    assert isinstance(
+        splat_source_metadata["dagster/source_paths"][DEFAULT_SOURCE_FILE_KEY],
+        SourcePathMetadataSet,
+    )
     AssetMaterialization(asset_key="a", metadata=splat_source_metadata)
 
-    assert dict(SourceDataMetadataSet()) == {"dagster/source_paths": []}
+    assert dict(SourceDataMetadataSet()) == {"dagster/source_paths": {}}
     assert SourceDataMetadataSet.extract(dict(SourceDataMetadataSet())) == SourceDataMetadataSet()