path control flow stuff to use the struct directly!
This commit is contained in:
@@ -15,22 +15,17 @@ from wf_core.models.conditions import (
|
||||
from wf_core.paths import GraphSourcePath
|
||||
|
||||
from .paths import GraphPath, context_path, input_path, state_path
|
||||
from .path_inputs import PathInput
|
||||
|
||||
|
||||
def _operand(value: object) -> PathOperand | LiteralOperand:
|
||||
if isinstance(value, PathExpr):
|
||||
return PathOperand(path=GraphSourcePath.parse(value.path))
|
||||
return PathOperand(path=value.source)
|
||||
if isinstance(value, GraphPath):
|
||||
return PathOperand(path=GraphSourcePath.parse(value.value))
|
||||
return PathOperand(path=value.path)
|
||||
return LiteralOperand(value=value)
|
||||
|
||||
|
||||
def _path_str(value: PathExpr | GraphPath) -> str:
|
||||
if isinstance(value, PathExpr):
|
||||
return value.path
|
||||
return value.value
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Expr:
|
||||
condition: Condition
|
||||
@@ -54,7 +49,11 @@ class Expr:
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PathExpr:
|
||||
path: str
|
||||
source: GraphSourcePath
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
return str(self.source)
|
||||
|
||||
def _binary(
|
||||
self, op: Literal["eq", "ne", "gt", "ge", "lt", "le"], other: object
|
||||
@@ -62,7 +61,7 @@ class PathExpr:
|
||||
return Expr(
|
||||
BinaryCondition(
|
||||
op=op,
|
||||
left=PathOperand(path=GraphSourcePath.parse(self.path)),
|
||||
left=PathOperand(path=self.source),
|
||||
right=_operand(other),
|
||||
)
|
||||
)
|
||||
@@ -111,25 +110,24 @@ class PathExpr:
|
||||
def expr(value: PathExpr | GraphPath) -> PathExpr:
|
||||
if isinstance(value, PathExpr):
|
||||
return value
|
||||
return PathExpr(path=value.value)
|
||||
return PathExpr(source=value.path)
|
||||
|
||||
|
||||
def state(field: str) -> PathExpr:
|
||||
return expr(state_path(field))
|
||||
def state(first: PathInput, *parts: object) -> PathExpr:
|
||||
return expr(state_path(first, *parts))
|
||||
|
||||
|
||||
def input(field: str) -> PathExpr:
|
||||
return expr(input_path(field))
|
||||
def input(first: PathInput, *parts: object) -> PathExpr:
|
||||
return expr(input_path(first, *parts))
|
||||
|
||||
|
||||
def context(field: str) -> PathExpr:
|
||||
return expr(context_path(field))
|
||||
def context(first: PathInput, *parts: object) -> PathExpr:
|
||||
return expr(context_path(first, *parts))
|
||||
|
||||
|
||||
def exists(value: PathExpr | GraphPath) -> Expr:
|
||||
return Expr(
|
||||
ExistsCondition(op="exists", path=GraphSourcePath.parse(_path_str(value)))
|
||||
)
|
||||
path = value.source if isinstance(value, PathExpr) else value.path
|
||||
return Expr(ExistsCondition(op="exists", path=path))
|
||||
|
||||
|
||||
def not_(value: Condition | Expr) -> Expr:
|
||||
|
||||
@@ -4,25 +4,40 @@ from collections.abc import Mapping
|
||||
from typing import TypeAlias
|
||||
|
||||
from .paths import GraphPath
|
||||
from .path_inputs import PathInput, coerce_graph_path, coerce_state_path
|
||||
|
||||
PathArg: TypeAlias = str | GraphPath
|
||||
PathArg: TypeAlias = PathInput | GraphPath
|
||||
|
||||
|
||||
def normalize_path(path: PathArg) -> str:
|
||||
"""Return display text for compatibility helpers.
|
||||
|
||||
Builder internals should prefer typed path normalizers. This function exists
|
||||
for older `bind_*` helpers and docs examples that still traffic in maps.
|
||||
"""
|
||||
if isinstance(path, GraphPath):
|
||||
return path.value
|
||||
return path
|
||||
return str(path)
|
||||
|
||||
|
||||
def bind_fields(**mapping: PathArg) -> dict[str, str]:
|
||||
return {
|
||||
normalize_path(source): destination for destination, source in mapping.items()
|
||||
str(coerce_graph_path(source.path if isinstance(source, GraphPath) else source)): (
|
||||
destination
|
||||
)
|
||||
for destination, source in mapping.items()
|
||||
}
|
||||
|
||||
|
||||
def bind_state(**mapping: PathArg) -> dict[str, str]:
|
||||
return {
|
||||
destination: normalize_path(target) for destination, target in mapping.items()
|
||||
destination: str(
|
||||
coerce_state_path(
|
||||
target.path if isinstance(target, GraphPath) else target,
|
||||
allow_legacy_root=True,
|
||||
)
|
||||
)
|
||||
for destination, target in mapping.items()
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable, Mapping
|
||||
import tomllib
|
||||
from typing import TypeAlias, cast
|
||||
|
||||
from wf_core.paths import GraphRoot, GraphSourcePath, LocalPath, StatePath
|
||||
|
||||
PathInput: TypeAlias = (
|
||||
str
|
||||
| Iterable[str]
|
||||
| Mapping[str, object]
|
||||
| GraphSourcePath
|
||||
| StatePath
|
||||
| LocalPath
|
||||
)
|
||||
|
||||
|
||||
def _parse_toml_key_expr(expr: str) -> tuple[str, ...]:
|
||||
"""Parse one authoring string as a TOML key expression.
|
||||
|
||||
We intentionally lean on `tomllib` instead of maintaining our own dotted-key
|
||||
parser. Quoted TOML keys are the escape hatch for literal dots and spaces.
|
||||
"""
|
||||
try:
|
||||
parsed = tomllib.loads(f"{expr} = true")
|
||||
except tomllib.TOMLDecodeError as exc:
|
||||
raise ValueError(
|
||||
f"invalid TOML key expression {expr!r}; use quoted keys, varargs, "
|
||||
"or an iterable for literal path segments"
|
||||
) from exc
|
||||
|
||||
parts: list[str] = []
|
||||
current: object = parsed
|
||||
while isinstance(current, dict):
|
||||
if len(current) != 1:
|
||||
raise ValueError(f"invalid TOML key expression {expr!r}")
|
||||
key, current = next(iter(current.items()))
|
||||
if not isinstance(key, str):
|
||||
raise ValueError(f"invalid TOML key expression {expr!r}")
|
||||
parts.append(key)
|
||||
if current is not True or not parts:
|
||||
raise ValueError(f"invalid TOML key expression {expr!r}")
|
||||
return tuple(parts)
|
||||
|
||||
|
||||
def _literal_parts(values: tuple[object, ...]) -> tuple[str, ...]:
|
||||
"""Normalize varargs or non-string iterables into literal path segments."""
|
||||
if not values:
|
||||
raise ValueError("expected at least one path segment")
|
||||
if len(values) == 1:
|
||||
value = values[0]
|
||||
if isinstance(value, str):
|
||||
return _parse_toml_key_expr(value)
|
||||
if isinstance(value, Iterable) and not isinstance(value, Mapping):
|
||||
parts = tuple(value)
|
||||
if all(isinstance(part, str) for part in parts):
|
||||
return cast(tuple[str, ...], parts)
|
||||
if all(isinstance(value, str) for value in values):
|
||||
return cast(tuple[str, ...], values)
|
||||
raise TypeError("expected a string, string iterable, or string varargs path")
|
||||
|
||||
|
||||
def _structural_parts(value: Mapping[str, object]) -> tuple[str, ...]:
|
||||
raw_parts = value.get("parts", [])
|
||||
if not isinstance(raw_parts, list) or not all(
|
||||
isinstance(part, str) for part in raw_parts
|
||||
):
|
||||
raise ValueError("expected structural path parts to be strings")
|
||||
return tuple(raw_parts)
|
||||
|
||||
|
||||
def coerce_graph_path(
|
||||
first: PathInput,
|
||||
*parts: object,
|
||||
root: GraphRoot | None = None,
|
||||
) -> GraphSourcePath:
|
||||
"""Coerce authoring input into a readable graph source path.
|
||||
|
||||
With an explicit root, strings are TOML key expressions and varargs /
|
||||
iterables are literal segments. Without a root, only existing structural or
|
||||
full graph paths are accepted so we do not infer roots from display text.
|
||||
"""
|
||||
if isinstance(first, GraphSourcePath):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to an existing graph path")
|
||||
if root is not None and first.root != root:
|
||||
raise ValueError(f"expected {root!r} graph path, got {first.root!r}")
|
||||
return first
|
||||
|
||||
if isinstance(first, StatePath):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to an existing state path")
|
||||
if root not in (None, "state"):
|
||||
raise ValueError(f"expected {root!r} graph path, got 'state'")
|
||||
return GraphSourcePath("state", first.parts)
|
||||
|
||||
if isinstance(first, Mapping):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to a structural graph path")
|
||||
graph_root = first.get("root")
|
||||
if graph_root not in GraphSourcePath._ROOTS:
|
||||
raise ValueError("expected structural graph source path")
|
||||
if root is not None and graph_root != root:
|
||||
raise ValueError(f"expected {root!r} graph path, got {graph_root!r}")
|
||||
return GraphSourcePath(cast(GraphRoot, graph_root), _structural_parts(first))
|
||||
|
||||
if root is None:
|
||||
if parts:
|
||||
raise TypeError("graph path varargs require an explicit root")
|
||||
if isinstance(first, str):
|
||||
return GraphSourcePath.parse(first)
|
||||
raise TypeError("expected graph path string or structural object")
|
||||
|
||||
return GraphSourcePath(root, _literal_parts((first, *parts)))
|
||||
|
||||
|
||||
def coerce_state_path(
|
||||
first: PathInput,
|
||||
*parts: object,
|
||||
allow_legacy_root: bool = False,
|
||||
) -> StatePath:
|
||||
"""Coerce authoring input into a writable workflow state path."""
|
||||
if isinstance(first, StatePath):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to an existing state path")
|
||||
return first
|
||||
if isinstance(first, GraphSourcePath):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to an existing graph path")
|
||||
if first.root != "state" or not first.parts:
|
||||
raise ValueError("expected state graph path")
|
||||
return StatePath(first.parts)
|
||||
if isinstance(first, Mapping):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to a structural state path")
|
||||
if first.get("root") != "state":
|
||||
raise ValueError("expected structural state path")
|
||||
return StatePath(_structural_parts(first))
|
||||
if allow_legacy_root and isinstance(first, str) and not parts:
|
||||
try:
|
||||
return StatePath.parse(first)
|
||||
except ValueError:
|
||||
pass
|
||||
return StatePath(_literal_parts((first, *parts)))
|
||||
|
||||
|
||||
def coerce_local_path(first: PathInput, *parts: object) -> LocalPath:
|
||||
"""Coerce authoring input into a node-local path."""
|
||||
if isinstance(first, LocalPath):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to an existing local path")
|
||||
return first
|
||||
if isinstance(first, Mapping):
|
||||
if parts:
|
||||
raise TypeError("cannot append path segments to a structural local path")
|
||||
if first.get("root") != "local":
|
||||
raise ValueError("expected structural local path")
|
||||
return LocalPath(_structural_parts(first))
|
||||
if isinstance(first, str) and first == "." and not parts:
|
||||
return LocalPath.root()
|
||||
return LocalPath(_literal_parts((first, *parts)))
|
||||
@@ -4,30 +4,41 @@ from dataclasses import dataclass
|
||||
|
||||
from wf_core.paths import GraphSourcePath
|
||||
|
||||
from .path_inputs import PathInput, coerce_graph_path
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class GraphPath:
|
||||
value: str
|
||||
path: GraphSourcePath
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
"""Validate authoring paths at construction so invalid roots fail early."""
|
||||
object.__setattr__(self, "value", str(GraphSourcePath.parse(self.value)))
|
||||
@property
|
||||
def value(self) -> str:
|
||||
"""Display compatibility for older authoring helpers."""
|
||||
return str(self.path)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.value
|
||||
|
||||
|
||||
def graph_path(value: str) -> GraphPath:
|
||||
return GraphPath(value)
|
||||
def graph_path(value: PathInput | GraphPath) -> GraphPath:
|
||||
if isinstance(value, GraphPath):
|
||||
return value
|
||||
return GraphPath(coerce_graph_path(value))
|
||||
|
||||
|
||||
def input_path(*parts: str) -> GraphPath:
|
||||
return GraphPath(str(GraphSourcePath.input(*parts)))
|
||||
def input_path(first: PathInput | None = None, *parts: object) -> GraphPath:
|
||||
if first is None:
|
||||
return GraphPath(GraphSourcePath("input"))
|
||||
return GraphPath(coerce_graph_path(first, *parts, root="input"))
|
||||
|
||||
|
||||
def state_path(*parts: str) -> GraphPath:
|
||||
return GraphPath(str(GraphSourcePath.state(*parts)))
|
||||
def state_path(first: PathInput | None = None, *parts: object) -> GraphPath:
|
||||
if first is None:
|
||||
return GraphPath(GraphSourcePath("state"))
|
||||
return GraphPath(coerce_graph_path(first, *parts, root="state"))
|
||||
|
||||
|
||||
def context_path(*parts: str) -> GraphPath:
|
||||
return GraphPath(str(GraphSourcePath.context(*parts)))
|
||||
def context_path(first: PathInput | None = None, *parts: object) -> GraphPath:
|
||||
if first is None:
|
||||
return GraphPath(GraphSourcePath("context"))
|
||||
return GraphPath(coerce_graph_path(first, *parts, root="context"))
|
||||
|
||||
Reference in New Issue
Block a user