lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: share expression traversal and history types

Browse Source
This commit is contained in:
lda
2026-09-10 02:16:38 +07:00 Verified
parent 93a6a03b9e
commit ed70223c1b
6 changed files with 60 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_core/models/input_bindings.py
+17 -1
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Iterator, Mapping
from typing import Annotated, Literal, TypeAliasType
from pydantic import BaseModel, ConfigDict, Field, field_validator
@@ -103,6 +103,22 @@ type InputExpression = Annotated[
]
def walk_expression_paths(
expression: InputExpression,
) -> Iterator[tuple[GraphSourcePath, str]]:
"""Yield graph path leaves with their stable expression locations."""
if isinstance(expression, PathExpression):
yield expression.path, "path"
elif isinstance(expression, ArrayExpression):
for index, item in enumerate(expression.items):
for path, suffix in walk_expression_paths(item):
yield path, f"items[{index}].{suffix}"
elif isinstance(expression, ObjectExpression):
for field, item in expression.fields.items():
for path, suffix in walk_expression_paths(item):
yield path, f"fields.{field}.{suffix}"
class OccurrenceExpression(BaseModel):
"""Reference one typed schedule-occurrence field.
src/wf_core/runtime/input_sources.py
+1 -21
View File
@@ -16,7 +16,7 @@ runtime values); budget parity there comes from the same model-level
from __future__ import annotations
from collections.abc import Iterator, Mapping, Sequence
from collections.abc import Mapping, Sequence
from typing import Any, Protocol
from wf_core.errors import WorkflowExecutionError
@@ -227,23 +227,3 @@ def resolve_schedule_input_bindings(
raise WorkflowExecutionError(f"{label} {location}: {exc}") from exc
return payload
def walk_expression_paths(
expression: InputExpression,
) -> Iterator[tuple[GraphSourcePath, str]]:
"""Yield ``(graph path, location)`` leaves in one shared traversal order.
Locations use the same ``items[i]`` / ``fields.name`` / ``path`` suffix
shape as :func:`wf_core.validation.steps._validate_expression_sources`,
so validation and runtime traverse leaves in the same order.
"""
if isinstance(expression, PathExpression):
yield expression.path, "path"
elif isinstance(expression, ArrayExpression):
for index, item in enumerate(expression.items):
for path, suffix in walk_expression_paths(item):
yield path, f"items[{index}].{suffix}"
elif isinstance(expression, ObjectExpression):
for field, item in expression.fields.items():
for path, suffix in walk_expression_paths(item):
yield path, f"fields.{field}.{suffix}"
src/wf_core/validation/steps.py
+6 -3
View File
@@ -10,7 +10,11 @@ from wf_core.models.conditions import (
PathOperand,
VariadicCondition,
)
from wf_core.models.input_bindings import InputExpression, InputExpressionBinding
from wf_core.models.input_bindings import (
InputExpression,
InputExpressionBinding,
walk_expression_paths,
)
from wf_core.models.schemas import NodeDef
from wf_core.models.steps import (
ConditionNode,
@@ -33,7 +37,6 @@ from wf_core.paths import (
is_valid_destination_path,
is_valid_source_path,
)
from wf_core.runtime.input_sources import walk_expression_paths
from wf_core.validation.issues import ValidationIssueCode, ValidationReport
@@ -247,7 +250,7 @@ def _validate_expression_sources(
"""Validate every graph path leaf while keeping one top-level target atomic.
Leaf traversal order is the canonical order defined by
:func:`wf_core.runtime.input_sources.walk_expression_paths`.
:func:`wf_core.models.input_bindings.walk_expression_paths`.
"""
for path, suffix in walk_expression_paths(expression):
_validate_source_path(
src/wf_scheduling/dispatch.py
+2 -2
View File
@@ -14,7 +14,7 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol, Union
from typing import Protocol
from wf_artifacts.runs.models import RunAdmission
from wf_core import RunState
@@ -32,7 +32,7 @@ class StillRunning:
"""Dispatched with an unknown outcome; the run stays admitted."""
DispatchResult = Union[Stopped, StillRunning]
DispatchResult = Stopped | StillRunning
class RunDispatcher(Protocol):
src/wf_scheduling/poll.py
+31 -22
View File
@@ -17,7 +17,7 @@ contains no fixture input, fixture environments, or canned results.
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
from typing import Any, cast
from wf_scheduling.calendar import (
CronSource,
@@ -31,7 +31,7 @@ from wf_scheduling.history import (
HistoryEntry,
HistoryRecorder,
)
from wf_scheduling.models import PendingCandidate
from wf_scheduling.models import OccurrenceKind, PendingCandidate
from wf_scheduling.ownership import (
SchedulerOwnership,
SecondOwnerError,
@@ -258,7 +258,7 @@ class Scheduler:
def _record(
self,
*,
kind: str,
kind: OccurrenceKind,
sched_id: str,
intended: datetime | None = None,
run_id: str | None = None,
@@ -275,7 +275,7 @@ class Scheduler:
self.history.record(
HistoryEntry(
schedule_id=sched_id,
kind=kind, # type: ignore[arg-type]
kind=kind,
resolved_at=intended,
run_id=run_id,
revision=revision,
@@ -504,13 +504,16 @@ class Scheduler:
run_id=run_id,
)
self.run_store.clear_executing(run_id)
kind = {
StoredRunStatus.COMPLETED: "completed",
StoredRunStatus.INTERRUPTED: "interrupted",
StoredRunStatus.FAILED: "failed",
}[stopped.status]
kind = cast(
OccurrenceKind,
{
StoredRunStatus.COMPLETED: "completed",
StoredRunStatus.INTERRUPTED: "interrupted",
StoredRunStatus.FAILED: "failed",
}[stopped.status],
)
self._record(
kind=kind, # type: ignore[arg-type]
kind=kind,
sched_id=admission.schedule_id,
intended=_admission_intended(self.run_store, run_id),
run_id=run_id,
@@ -563,13 +566,16 @@ class Scheduler:
run_id=run_id,
)
self.run_store.clear_executing(run_id)
kind = {
StoredRunStatus.COMPLETED: "completed",
StoredRunStatus.INTERRUPTED: "interrupted",
StoredRunStatus.FAILED: "failed",
}[stopped.status]
kind = cast(
OccurrenceKind,
{
StoredRunStatus.COMPLETED: "completed",
StoredRunStatus.INTERRUPTED: "interrupted",
StoredRunStatus.FAILED: "failed",
}[stopped.status],
)
self._record(
kind=kind, # type: ignore[arg-type]
kind=kind,
sched_id=admission.schedule_id,
intended=_admission_intended(self.run_store, run_id),
run_id=run_id,
@@ -597,11 +603,14 @@ class Scheduler:
history. Returns whether an entry was appended.
"""
self._require_ownership()
kind = {
"completed": "completed",
"interrupted": "interrupted",
"failed": "failed",
}.get(status_value)
kind = cast(
OccurrenceKind | None,
{
"completed": "completed",
"interrupted": "interrupted",
"failed": "failed",
}.get(status_value),
)
if kind is None:
return False
try:
@@ -614,7 +623,7 @@ class Scheduler:
if self.history.has_terminal(sched_id, run_id, kind, checkpoint_id):
return False
self._record(
kind=kind, # type: ignore[arg-type]
kind=kind,
sched_id=sched_id,
intended=admission.scheduled_at,
run_id=run_id,
src/wf_scheduling/recovery.py
+3 -2
View File
@@ -62,6 +62,7 @@ from wf_scheduling.history import (
HistoryEntry,
HistoryRecorder,
)
from wf_scheduling.models import OccurrenceKind
from wf_scheduling.ownership import (
SchedulerOwnership,
SecondOwnerError,
@@ -676,7 +677,7 @@ def _reconcile_terminal(
history: HistoryRecorder,
sched_id: str | None,
run_id: str,
kind: str,
kind: OccurrenceKind,
checkpoint_id: str | None,
intended: datetime | None,
revision: int | None,
@@ -697,7 +698,7 @@ def _reconcile_terminal(
history.record(
HistoryEntry(
schedule_id=sched_id,
kind=kind, # type: ignore[arg-type]
kind=kind,
resolved_at=intended,
run_id=run_id,
revision=revision,