feedback

Author: salazarm

python_modules/dagster/dagster/_core/definitions/asset_selection.py

@@ -204,6 +204,30 @@ def key_prefixes(
             include_sources=include_sources,
         )
 
+    @staticmethod
+    @beta_param(param="include_sources")
+    def key_substring(
+        key_substring: str, include_sources: bool = False
+    ) -> "KeySubstringAssetSelection":
+        """Returns a selection that includes assets whose string representation contains the provided substring and all the asset checks that target it.
+
+        Args:
+            include_sources (bool): If True, then include external assets matching the substring
+                in the selection.
+
+        Examples:
+            .. code-block:: python
+              # match any asset key containing "bc"
+              # e.g. AssetKey(["a", "bcd"]) would match, but not AssetKey(["ab", "cd"]).
+              AssetSelection.key_substring("bc")
+              # match any asset key containing "b/c"
+              # e.g. AssetKey(["ab", "cd"]) would match.
+              AssetSelection.key_substring("b/c")
+        """
+        return KeySubstringAssetSelection(
+            selected_key_substring=key_substring, include_sources=include_sources
+        )
+
     @public
     @staticmethod
     @beta_param(param="include_sources")
@@ -1172,6 +1196,29 @@ def to_serializable_asset_selection(self, asset_graph: BaseAssetGraph) -> "Asset
         return self
 
 
+@whitelist_for_serdes
+@record
+class KeySubstringAssetSelection(AssetSelection):
+    selected_key_substring: str
+    include_sources: bool
+
+    def resolve_inner(
+        self, asset_graph: BaseAssetGraph, allow_missing: bool
+    ) -> AbstractSet[AssetKey]:
+        base_set = (
+            asset_graph.get_all_asset_keys()
+            if self.include_sources
+            else asset_graph.materializable_asset_keys
+        )
+        return {key for key in base_set if self.selected_key_substring in key.to_user_string()}
+
+    def to_serializable_asset_selection(self, asset_graph: BaseAssetGraph) -> "AssetSelection":
+        return self
+
+    def to_selection_str(self) -> str:
+        return f'key_substring:"{self.selected_key_substring}"'
+
+
 @whitelist_for_serdes
 @record
 class KeyWildCardAssetSelection(AssetSelection):

python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_selection.py

@@ -211,6 +211,22 @@ def test_asset_selection_key_prefixes(all_assets: _AssetList):
     assert sel.resolve(all_assets) == {earth.key}
 
 
+def test_asset_selection_key_substring(all_assets: _AssetList):
+    sel = AssetSelection.key_substring("alice")
+    assert sel.resolve(all_assets) == _asset_keys_of({alice})
+
+    sel = AssetSelection.key_substring("ls/ze")
+    assert sel.resolve(all_assets) == _asset_keys_of({zebra})
+
+    # does not include source assets by default
+    sel = AssetSelection.key_substring("celestial")
+    assert sel.resolve(all_assets) == set()
+
+    # includes source assets if flag set
+    sel = AssetSelection.key_substring("celestial/e", include_sources=True)
+    assert sel.resolve(all_assets) == {earth.key}
+
+
 def test_select_source_asset_keys():
     a = SourceAsset("a")
     selection = AssetSelection.keys(a.key)