Skip to content

Commit

Permalink
support prefixed asset keys in blueprints, via `AssetKey.from_user_st…
Browse files Browse the repository at this point in the history
…ring` (dagster-io#23016)

## Summary & Motivation

## How I Tested These Changes
  • Loading branch information
sryza authored Jul 15, 2024
1 parent 8ffc088 commit 515d030
Show file tree
Hide file tree
Showing 2 changed files with 16 additions and 1 deletion.
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from abc import abstractmethod
from typing import AbstractSet, Any, Mapping, Optional, Sequence

from dagster._core.definitions.asset_key import AssetKey
from dagster._core.definitions.asset_spec import AssetSpec
from dagster._core.definitions.assets import unique_id_from_asset_and_check_keys
from dagster._core.definitions.decorators.asset_decorator import multi_asset
Expand All @@ -22,7 +23,12 @@ class AssetSpecModel(DagsterModel):
tags: Mapping[str, str] = {}

def to_asset_spec(self) -> AssetSpec:
return AssetSpec(**self.__dict__)
return AssetSpec(
**{
**self.__dict__,
"key": AssetKey.from_user_string(self.key),
},
)


class BlueprintAssetsDefinition(Blueprint):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,15 @@ def test_single_asset_shell_command_blueprint() -> None:
).success


def test_single_asset_shell_command_blueprint_key_prefix() -> None:
single_asset_blueprint = ShellCommandBlueprint(
assets=[AssetSpecModel(key="prefix/asset1")], command=["echo", '"hello"']
)
defs = single_asset_blueprint.build_defs()
asset1 = cast(AssetsDefinition, next(iter(defs.assets)))
assert asset1.key == AssetKey(["prefix", "asset1"])


def test_single_asset_shell_command_blueprint_str_command() -> None:
single_asset_blueprint = ShellCommandBlueprint(
assets=[AssetSpecModel(key="asset1")], command='echo "hello world"'
Expand Down

0 comments on commit 515d030

Please sign in to comment.