Appointat commented on code in PR #737:
URL: https://github.com/apache/geaflow/pull/737#discussion_r2791968305
##########
geaflow-ai/src/operator/casts/casts/core/gremlin_state.py:
##########
@@ -0,0 +1,265 @@
+"""Gremlin traversal state machine for validating graph traversal steps."""
+
+from dataclasses import dataclass
+from typing import Literal, Sequence, TypedDict
+
+from casts.core.interfaces import GraphSchema
+
+
+GremlinState = Literal["V", "E", "P", "END"]
+
+
+class GremlinStateDefinition(TypedDict):
+ """Typed representation of a Gremlin state definition."""
+
+ options: list[str]
+ transitions: dict[str, GremlinState]
+
+
+# Gremlin Step State Machine
+# Defines valid transitions between step types (V: Vertex, E: Edge, P:
Property)
+GREMLIN_STEP_STATE_MACHINE: dict[GremlinState, GremlinStateDefinition] = {
+ # State: current element is a Vertex
+ "V": {
+ "options": [
+ "out('label')",
+ "in('label')",
+ "both('label')",
+ "outE('label')",
+ "inE('label')",
+ "bothE('label')",
+ "has('prop','value')",
+ "dedup()",
+ "simplePath()",
+ "order().by('prop')",
+ "limit(n)",
+ "values('prop')",
+ "stop",
+ ],
+ "transitions": {
+ "out": "V",
+ "in": "V",
+ "both": "V",
+ "outE": "E",
+ "inE": "E",
+ "bothE": "E",
+ "has": "V",
+ "dedup": "V",
+ "simplePath": "V",
+ "order": "V",
+ "limit": "V",
+ "values": "P",
+ "stop": "END",
+ },
+ },
+ # State: current element is an Edge
+ "E": {
+ "options": [
+ "inV()",
+ "outV()",
+ "otherV()",
+ "has('prop','value')",
+ "dedup()",
+ "simplePath()",
+ "order().by('prop')",
+ "limit(n)",
+ "values('prop')",
+ "stop",
+ ],
+ "transitions": {
+ "inV": "V",
+ "outV": "V",
+ "otherV": "V",
+ "has": "E",
+ "dedup": "E",
+ "simplePath": "E",
+ "order": "E",
+ "limit": "E",
+ "values": "P",
+ "stop": "END",
+ },
+ },
+ # State: current element is a Property/Value
+ "P": {
+ "options": ["order()", "limit(n)", "dedup()", "simplePath()", "stop"],
+ "transitions": {
+ "order": "P",
+ "limit": "P",
+ "dedup": "P",
+ "simplePath": "P",
+ "stop": "END",
+ },
+ },
+ "END": {"options": [], "transitions": {}},
+}
+
+_MODIFIER_STEPS = {"by"}
+_MODIFIER_COMPATIBILITY = {"by": {"order"}}
+
+
+@dataclass(frozen=True)
+class ParsedStep:
+ """Parsed step representation for traversal signatures."""
+
+ raw: str
+ name: str
+
+
+def _normalize_signature(signature: str) -> str:
+ """Normalize a traversal signature by stripping the V() prefix and
separators."""
+ normalized = signature.strip()
+ if not normalized or normalized == "V()":
+ return ""
+
+ if normalized.startswith("V()"):
+ normalized = normalized[3:]
+ elif normalized.startswith("V"):
+ normalized = normalized[1:]
+
+ return normalized.lstrip(".")
+
+
+def _split_steps(signature: str) -> list[str]:
+ """Split a traversal signature into raw step segments."""
+ if not signature:
+ return []
+
+ steps: list[str] = []
+ current: list[str] = []
+ depth = 0
+
+ for ch in signature:
+ if ch == "." and depth == 0:
+ if current:
+ steps.append("".join(current))
+ current = []
+ continue
+
+ if ch == "(":
+ depth += 1
+ elif ch == ")":
+ depth = max(depth - 1, 0)
+
+ current.append(ch)
+
+ if current:
+ steps.append("".join(current))
+
+ return [step for step in steps if step]
+
+
+def _extract_step_name(step: str) -> str:
+ """Extract the primary step name from a step string."""
+ head = step.split("(", 1)[0]
+ if "." in head:
+ return head.split(".", 1)[0]
+ return head
+
+
+def _combine_modifiers(steps: Sequence[str]) -> list[str]:
+ """Combine modifier steps (e.g., order().by()) into a single step
string."""
+ combined: list[str] = []
+ for step in steps:
+ step_name = _extract_step_name(step)
+ if step_name in _MODIFIER_STEPS and combined:
+ previous_name = _extract_step_name(combined[-1])
+ if previous_name in _MODIFIER_COMPATIBILITY.get(step_name, set()):
+ combined[-1] = f"{combined[-1]}.{step}"
+ continue
+ combined.append(step)
+ return combined
+
+
+def _parse_traversal_signature(signature: str) -> list[ParsedStep]:
+ """Parse traversal signature into steps with normalized names."""
+ normalized = _normalize_signature(signature)
+ raw_steps = _combine_modifiers(_split_steps(normalized))
+ return [ParsedStep(raw=step, name=_extract_step_name(step)) for step in
raw_steps]
+
+
+class GremlinStateMachine:
+ """State machine for validating Gremlin traversal steps and determining
next valid options."""
+
+ @staticmethod
+ def parse_traversal_signature(structural_signature: str) -> list[str]:
+ """Parse traversal signature into decision steps for display or
history."""
+ return [step.raw for step in
_parse_traversal_signature(structural_signature)]
+
+ @staticmethod
+ def get_state_and_options(
+ structural_signature: str, graph_schema: GraphSchema, node_id: str
Review Comment:
For backward compatibility, memory can be independent of a fixed schema, so
we assume the schema is dynamic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
