introduce dagster/storage_kind tag

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

@@ -1 +1,13 @@
+from typing import Optional
+
 from .tag_set import NamespacedTagSet as NamespacedTagSet
+
+
+class StorageKindTagSet(NamespacedTagSet):
+    """Metadata entries which describe how an asset is stored."""
+
+    storage_kind: Optional[str]
+
+    @classmethod
+    def namespace(cls) -> str:
+        return "dagster"

python_modules/dagster/dagster_tests/asset_defs_tests/test_assets.py

@@ -43,6 +43,7 @@
 from dagster._core.definitions.decorators.asset_decorator import graph_asset
 from dagster._core.definitions.events import AssetMaterialization
 from dagster._core.definitions.result import MaterializeResult
+from dagster._core.definitions.tags import StorageKindTagSet
 from dagster._core.errors import (
     DagsterInvalidDefinitionError,
     DagsterInvalidInvocationError,
@@ -2146,6 +2147,21 @@ def asset1(): ...
         def asset2(): ...
 
 
+def test_asset_with_storage_kind_tag() -> None:
+    @asset(tags={**StorageKindTagSet(storage_kind="snowflake")})
+    def asset1(): ...
+
+    assert asset1.tags_by_key[asset1.key] == {"dagster/storage_kind": "snowflake"}
+
+    @asset(tags={**StorageKindTagSet(storage_kind="snowflake"), "a": "b"})
+    def asset2(): ...
+
+    assert asset2.tags_by_key[asset2.key] == {
+        "dagster/storage_kind": "snowflake",
+        "a": "b",
+    }
+
+
 def test_asset_spec_with_tags():
     @multi_asset(specs=[AssetSpec("asset1", tags={"a": "b"})])
     def assets(): ...