tag no value

branch-name: tag-no-value

python_modules/dagster/dagster/__init__.py

@@ -559,6 +559,7 @@
 from dagster._core.storage.tags import (
     MAX_RUNTIME_SECONDS_TAG as MAX_RUNTIME_SECONDS_TAG,
     MEMOIZED_RUN_TAG as MEMOIZED_RUN_TAG,
+    TAG_NO_VALUE as TAG_NO_VALUE,
 )
 from dagster._core.storage.upath_io_manager import UPathIOManager as UPathIOManager
 from dagster._core.types.config_schema import (

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

@@ -19,6 +19,7 @@
     fetch_sources,
     parse_clause,
 )
+from dagster._core.storage.tags import TAG_NO_VALUE
 from dagster._serdes.serdes import whitelist_for_serdes
 
 from .asset_check_spec import AssetCheckKey
@@ -195,8 +196,16 @@ def tag_string(cls, string: str, include_sources: bool = False) -> "AssetSelecti
             include_sources (bool): If True, then include source assets matching the group in the
                 selection.
         """
-        key, value = string.split("=")
-        return TagAssetSelection(key=key, value=value, include_sources=include_sources)
+        split_by_equals_segments = string.split("=")
+        if len(split_by_equals_segments) == 1:
+            return TagAssetSelection(
+                key=string, value=TAG_NO_VALUE, include_sources=include_sources
+            )
+        elif len(split_by_equals_segments) == 2:
+            key, value = split_by_equals_segments
+            return TagAssetSelection(key=key, value=value, include_sources=include_sources)
+        else:
+            check.failed(f"Invalid tag selection string: {string}. Must have no more than one '='.")
 
     @public
     @staticmethod

python_modules/dagster/dagster/_core/storage/tags.py

@@ -85,6 +85,9 @@
 ]
 
 
+TAG_NO_VALUE = "__dagster_novalue"
+
+
 class TagType(Enum):
     # Custom tag provided by a user
     USER_PROVIDED = "USER_PROVIDED"

python_modules/dagster/dagster_tests/asset_defs_tests/test_asset_selection.py

@@ -5,6 +5,7 @@
 
 import pytest
 from dagster import (
+    TAG_NO_VALUE,
     AssetIn,
     AssetOut,
     AssetSpec,
@@ -711,6 +712,7 @@ def test_deserialize_old_all_asset_selection():
 
 def test_from_string_tag():
     assert AssetSelection.from_string("tag:foo=bar") == AssetSelection.tag("foo", "bar")
+    assert AssetSelection.from_string("tag:foo") == AssetSelection.tag("foo", TAG_NO_VALUE)
 
 
 def test_tag():
@@ -736,10 +738,18 @@ def test_tag_string():
             AssetSpec("asset2", tags={"foo": "fooval2"}),
             AssetSpec("asset3", tags={"foo": "fooval", "bar": "barval"}),
             AssetSpec("asset4", tags={"bar": "barval"}),
+            AssetSpec("asset5", tags={"baz": TAG_NO_VALUE}),
+            AssetSpec("asset6", tags={"baz": TAG_NO_VALUE, "bar": "barval"}),
         ]
     )
     def assets(): ...
 
     assert AssetSelection.tag_string("foo=fooval").resolve([assets]) == {
-        AssetKey(k) for k in ["asset1", "asset3"]
+        AssetKey("asset1"),
+        AssetKey("asset3"),
+    }
+    assert AssetSelection.tag_string("foo").resolve([assets]) == set()
+    assert AssetSelection.tag_string("baz").resolve([assets]) == {
+        AssetKey("asset5"),
+        AssetKey("asset6"),
     }