diff --git a/CLAUDE.md b/CLAUDE.md index 9843d55..36a9932 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -92,6 +92,10 @@ The critical execution order is: `main_event_loop` → `microstep` → `main_eve `setup(guards=..., actions=..., delays=..., actors=...).create_machine(config)` - **Snapshot serialization** (0.6.0, `from xstate import serialize_snapshot, deserialize_snapshot`) — persist and restore State; `create_actor(machine, snapshot=...)` for round-trip replay +- **State tags** (0.7.0) — declare `tags: ["loading"]` (or a single string) on any state node; + query the snapshot with `state.has_tag("loading")` / `state.hasTag(...)` or read the aggregated + `state.tags` frozenset. Tags union across the whole active configuration (compound ancestors + + parallel regions) and are recomputed from the machine definition, so snapshots stay tag-free Handler-signature note: guards/assigners are invoked arity-aware by `algorithm._invoke`, which supports four calling conventions: `()`, `(context)`, diff --git a/src/xstate/config_parser.py b/src/xstate/config_parser.py index 74fac6b..e205962 100644 --- a/src/xstate/config_parser.py +++ b/src/xstate/config_parser.py @@ -105,10 +105,28 @@ def _resolve_node(self, node: StateNode, *, path: str) -> None: path=self._path(path, "target"), ) + node.tags = self._build_tags(config.get("tags"), self._path(path, "tags")) + node.donedata = self._build_output(node, path) self._build_configured_transitions(node, path) self._build_initial_transition(node, path) + def _build_tags(self, tags_config: Any, path: str) -> tuple[str, ...]: + if tags_config is None: + return () + if isinstance(tags_config, str): + return (tags_config,) + if isinstance(tags_config, (list, tuple)): + if not all(isinstance(t, str) for t in tags_config): + raise InvalidConfigError( + f"{path}: every tag must be a string, got {tags_config!r}." + ) + return tuple(tags_config) + raise InvalidConfigError( + f"{path}: 'tags' must be a string or a list of strings, " + f"got {type(tags_config)!r}." + ) + def _build_output(self, node: StateNode, path: str) -> Any: if node.type != "final": return None diff --git a/src/xstate/schema.py b/src/xstate/schema.py index ae35b1b..ab8ac1e 100644 --- a/src/xstate/schema.py +++ b/src/xstate/schema.py @@ -61,6 +61,7 @@ class StateNodeConfig(TypedDict, total=False): target: TransitionTarget output: Any data: Any + tags: str | list[str] class MachineConfig(StateNodeConfig, total=False): diff --git a/src/xstate/state.py b/src/xstate/state.py index afb2269..94e1bc2 100644 --- a/src/xstate/state.py +++ b/src/xstate/state.py @@ -72,6 +72,25 @@ def __init__( self.status = "active" self.output = None + @property + def tags(self) -> frozenset[str]: + """Union of the ``tags`` declared on every active state node. + + XState v5 surfaces ``snapshot.tags`` as the set of tags across the + current configuration; querying it is the idiomatic way to ask "is the + machine loading / busy / editable" without enumerating state values. + """ + return frozenset( + tag for node in self.configuration for tag in node.tags + ) + + def has_tag(self, tag: str) -> bool: + """Return True if any active state node declares *tag* (v5 ``hasTag``).""" + return any(tag in node.tags for node in self.configuration) + + # XState v5 spells this ``hasTag``; expose both for JS-parity ergonomics. + hasTag = has_tag + def can(self, event: Any) -> bool: """Return True if any enabled transition exists for *event* right now. diff --git a/src/xstate/state_node.py b/src/xstate/state_node.py index eed5bcf..6f95e2b 100644 --- a/src/xstate/state_node.py +++ b/src/xstate/state_node.py @@ -36,6 +36,7 @@ class StateNode: after: list[tuple[Any, str]] = field(default_factory=list) invoke: list[dict[str, Any]] = field(default_factory=list) initial_transition: Transition | None = None + tags: tuple[str, ...] = field(default_factory=tuple) @property def history_states(self) -> list[StateNode]: diff --git a/tests/test_tags.py b/tests/test_tags.py new file mode 100644 index 0000000..f676b64 --- /dev/null +++ b/tests/test_tags.py @@ -0,0 +1,203 @@ +"""Tests for state tags + has_tag()/hasTag() (0.7.0). + +XState v5 lets a state declare ``tags: ["loading"]`` and query the running +snapshot with ``state.hasTag("loading")``. Tags aggregate across the whole +active configuration (compound ancestors + parallel regions). +""" + +import pytest + +from xstate import Machine, create_actor +from xstate.exceptions import InvalidConfigError + +# --------------------------------------------------------------------------- +# Parsing +# --------------------------------------------------------------------------- + + +def test_tags_as_list(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": { + "loading": {"tags": ["busy", "network"]}, + "idle": {}, + }, + } + ) + assert machine.initial_state.tags == frozenset({"busy", "network"}) + + +def test_tags_as_single_string(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": { + "loading": {"tags": "busy"}, + "idle": {}, + }, + } + ) + assert machine.initial_state.tags == frozenset({"busy"}) + + +def test_no_tags_is_empty_frozenset(): + machine = Machine( + {"id": "m", "initial": "idle", "states": {"idle": {}}} + ) + assert machine.initial_state.tags == frozenset() + + +def test_non_string_tag_raises(): + with pytest.raises(InvalidConfigError, match="every tag must be a string"): + Machine( + { + "id": "m", + "initial": "a", + "states": {"a": {"tags": ["ok", 123]}}, + } + ) + + +def test_invalid_tags_type_raises(): + with pytest.raises(InvalidConfigError, match="must be a string or a list"): + Machine( + { + "id": "m", + "initial": "a", + "states": {"a": {"tags": {"not": "valid"}}}, + } + ) + + +# --------------------------------------------------------------------------- +# has_tag / hasTag query +# --------------------------------------------------------------------------- + + +def test_has_tag_true(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": {"loading": {"tags": ["busy"]}, "idle": {}}, + } + ) + assert machine.initial_state.has_tag("busy") is True + + +def test_has_tag_false(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": {"loading": {"tags": ["busy"]}, "idle": {}}, + } + ) + assert machine.initial_state.has_tag("idle") is False + + +def test_hasTag_camelcase_alias(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": {"loading": {"tags": ["busy"]}, "idle": {}}, + } + ) + assert machine.initial_state.hasTag("busy") is True + assert machine.initial_state.hasTag("nope") is False + + +# --------------------------------------------------------------------------- +# Tags update across transitions +# --------------------------------------------------------------------------- + + +def test_tags_change_on_transition(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": { + "loading": {"tags": ["busy"], "on": {"DONE": "ready"}}, + "ready": {"tags": ["interactive"]}, + }, + } + ) + state = machine.initial_state + assert state.has_tag("busy") + + state = machine.transition(state, "DONE") + assert state.has_tag("interactive") + assert not state.has_tag("busy") + + +def test_tags_via_actor(): + machine = Machine( + { + "id": "m", + "initial": "loading", + "states": { + "loading": {"tags": ["busy"], "on": {"DONE": "ready"}}, + "ready": {}, + }, + } + ) + actor = create_actor(machine).start() + assert actor.get_snapshot().has_tag("busy") + actor.send("DONE") + assert not actor.get_snapshot().has_tag("busy") + + +# --------------------------------------------------------------------------- +# Aggregation across compound ancestors and parallel regions +# --------------------------------------------------------------------------- + + +def test_tags_aggregate_from_compound_ancestor(): + machine = Machine( + { + "id": "m", + "initial": "parent", + "states": { + "parent": { + "tags": ["outer"], + "initial": "child", + "states": { + "child": {"tags": ["inner"]}, + }, + }, + }, + } + ) + state = machine.initial_state + # Both the active compound ancestor and the leaf contribute tags. + assert state.tags == frozenset({"outer", "inner"}) + assert state.has_tag("outer") + assert state.has_tag("inner") + + +def test_tags_aggregate_across_parallel_regions(): + machine = Machine( + { + "id": "m", + "type": "parallel", + "states": { + "a": { + "initial": "a1", + "states": {"a1": {"tags": ["region-a"]}}, + }, + "b": { + "initial": "b1", + "states": {"b1": {"tags": ["region-b"]}}, + }, + }, + } + ) + state = machine.initial_state + assert state.has_tag("region-a") + assert state.has_tag("region-b") + assert {"region-a", "region-b"} <= state.tags