Sync only the declaring module before a run; skip git-ignored paths

`fluksio run` walked the working directory and imported every module in
it, which ran a research repository's analysis scripts before the run was
submitted. A flow's origin now records the file and module name that
declared it, and run/sweep import just that, under the same name so the
generated node bodies do not change. A flow with no record still walks,
now skipping what .gitignore covers.

Every code-defined flow reports `flow` changed once on its next sync,
since the origin gained two fields.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QgG6TJw5qBZzaLghQSRVrq
This commit is contained in:
2026-09-16 16:55:02 +02:00
co-authored by Claude Opus 5
parent 7dfe646e2b
commit fccf89f849
9 changed files with 246 additions and 22 deletions
+6
View File
@@ -253,6 +253,12 @@ class FlowOrigin(BaseModel):
#: a run stamped with a dirty commit names code that was never stored.
commit: str = ""
dirty: bool = False
#: The file that declared the flow, relative to ``repo``, and the module
#: name it was imported under. What ``fluksio run`` imports by default,
#: instead of every script below the working directory — by that name, or
#: the generated node bodies would change with how the flow was synced.
declared_in: str = ""
module: str = ""
class FlowInput(BaseModel):
+37 -1
View File
@@ -582,6 +582,35 @@ def _code_of(spec: NodeSpec) -> dict[str, Any]:
return {"code_files": files, "code_digest": digest_of(files)}
def _declared_at() -> tuple[str, str]:
"""The file and module name of whatever constructed the flow.
Empty for a declaration no import could reach again: a script run
directly, or a notebook cell, which is not a file on disk.
"""
frame = inspect.currentframe()
# Out of this helper, then out of `Flow.__init__`.
for _ in range(2):
frame = frame.f_back if frame is not None else None
if frame is None:
return "", ""
module = str(frame.f_globals.get("__name__") or "")
path = Path(frame.f_code.co_filename)
if module == "__main__" or not path.is_file():
return "", ""
return str(path.resolve()), module
def _relative(path: str, repo: str) -> str:
"""``path`` as seen from ``repo``, or empty when it is not inside it."""
if not path or not repo:
return ""
try:
return Path(path).relative_to(Path(repo).resolve()).as_posix()
except ValueError:
return ""
def _serialisable(value: Any) -> bool:
try:
json.dumps(value)
@@ -619,6 +648,9 @@ class Flow:
self.inputs = list(inputs)
self.outputs = list(outputs)
self.nodes = [_spec_of(entry) for entry in nodes]
#: Where this flow was declared: the file, and the module name it was
#: imported under. What `fluksio run` imports by default.
self.declared_in, self.module = _declared_at()
self._resolve_types()
self._validate()
self._nodes = [_check(spec, mode) for spec in self.nodes]
@@ -719,7 +751,11 @@ class Flow:
"outputs": self.outputs,
}
if origin is not None:
doc["origin"] = origin
declared_in = _relative(self.declared_in, str(origin.get("repo") or ""))
doc["origin"] = origin | {
"declared_in": declared_in,
"module": self.module if declared_in else "",
}
return doc
def shims(self) -> dict[str, str]:
+60 -14
View File
@@ -35,6 +35,7 @@ from fluksio.sdk.client import (
Client,
RunHandle,
data_dir,
ignored,
login,
origin_of,
repo_root,
@@ -152,22 +153,28 @@ def _walkable(entry: Path) -> bool:
)
def _below(root: Path) -> Iterator[Path]:
def _below(root: Path, skip: frozenset[Path] = frozenset()) -> Iterator[Path]:
"""Every module and package under a plain directory, however deep.
One directory per study — `dev/s1_baseline/study.py` — is a layout people
have, and naming each of them on the command line is bookkeeping the tool
can do. A package is yielded whole and not descended into: its own walk
imports its submodules under the right names.
``skip`` is what the repository ignores. Importing a module runs it, and
what a repository ignores — `mlruns/`, `results/` — is output rather than
studies, and often large enough that walking it is most of the cost.
"""
for entry in sorted(root.iterdir()):
if entry in skip:
continue
if entry.is_file() and entry.suffix == ".py":
yield entry
elif _walkable(entry):
if (entry / "__init__.py").exists():
yield entry
else:
yield from _below(entry)
yield from _below(entry, skip)
def _import_package(directory: Path) -> None:
@@ -207,7 +214,7 @@ def discover(targets: list[str], keep_going: bool = False) -> list[Flow]:
# A plain directory — a repository root, or a directory of studies.
# Its own modules, the packages inside it, and the same again all the
# way down, so one folder per study needs no naming.
for entry in _below(path):
for entry in _below(path, ignored(path)):
try:
if entry.is_dir():
_import_package(entry)
@@ -496,7 +503,32 @@ def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]:
return params
def _sync_first(client: Client, targets: list[str] | None = None) -> None:
def _declaration(client: Client, flow: str) -> tuple[str, str, Path] | None:
"""Where a flow's last sync found it: import root, module name, file.
None when there is no such record to go on: a flow never synced, one drawn
on the canvas, an engine older than the field, or a checkout where that
file is not where the name says.
"""
if not flow:
return None
stored = client.get_flow(flow) or {}
origin = (stored.get("definition") or {}).get("origin") or {}
declared_in = str(origin.get("declared_in") or "")
module = str(origin.get("module") or "")
if not declared_in or not module:
return None
path = Path(repo_root(".")) / declared_in
stem = path.parent if path.name == "__init__.py" else path.with_suffix("")
parts = module.split(".")
if not path.is_file() or list(stem.parts[-len(parts) :]) != parts:
return None
return str(stem.parents[len(parts) - 1]), module, path
def _sync_first(
client: Client, targets: list[str] | None = None, flow: str = ""
) -> None:
"""Upload what the working directory declares, before running it.
The reason a run exists is usually the edit that came before it, and
@@ -504,24 +536,38 @@ def _sync_first(client: Client, targets: list[str] | None = None) -> None:
done. So `run` syncs by default — including the worker refresh, which is
what makes an edit to your own package take effect at all.
The whole directory, downwards: from a repository root the flow is usually
in a study folder below, and the alternative is syncing by hand and then
running with `--no-sync`. The upload is already a no-op for a flow nothing
changed in, so the cost is importing the other studies — which `--sync`
narrows when that is not free.
What is synced is the module that declared the flow, as the last sync
recorded it. Importing a module runs it, so walking the working directory
would run every analysis script in it before the run was even submitted.
A flow with no such record — never synced, or an engine older than the
field — falls back to the whole directory, downwards, skipping what the
repository ignores; a study there that will not import is a warning.
`--sync` names the targets outright.
A directory that declares nothing is not an error: a flow drawn on the
canvas is run the same way, and has nothing to upload.
"""
declared = None if targets else _declaration(client, flow)
try:
flows = discover(targets or ["."], keep_going=True)
except (ImportError, SyncError) as exc:
if declared is None:
flows = discover(targets or ["."], keep_going=True)
else:
# Under the name it was synced with, so the node bodies it
# generates are the ones already stored.
_import(*declared)
flows = list(FLOWS.values())
except SyncError as exc:
# Do not fail a run for a module the run may not even need.
_say(f"warning: nothing synced — {exc}")
return
except Exception as exc:
# A module named outright is imported without the walk's forgiveness,
# and importing it runs it: this is whatever it raised at its top.
_say(f"warning: nothing synced — {type(exc).__name__}: {exc}")
return
if not flows:
return
repo = repo_root((targets or ["."])[0])
repo = repo_root(declared[2] if declared else (targets or ["."])[0])
reports = sync(flows, client, origin=origin_of(repo))
changed = [r for r in reports if not r.unchanged]
if changed:
@@ -606,7 +652,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int:
try:
with _client_for(args) as client:
if not args.no_sync:
_sync_first(client, args.sync)
_sync_first(client, args.sync, args.flow)
stored = client.get_flow(args.flow)
if stored is None:
return _fail(f"no flow '{args.flow}' on that engine")
@@ -1145,7 +1191,7 @@ def cmd_sweep(args: argparse.Namespace) -> int:
try:
with _client_for(args) as client:
if not args.no_sync:
_sync_first(client, args.sync)
_sync_first(client, args.sync, args.flow)
stored = client.get_flow(args.flow)
if stored is None:
return _fail(f"no flow '{args.flow}' on that engine")
+20
View File
@@ -708,6 +708,26 @@ def repo_root(start: str | Path) -> str:
return found or str(directory)
def ignored(directory: Path) -> frozenset[Path]:
"""What the repository ignores below a directory, as git itself reads it.
Directories come back whole rather than file by file, so an ignored
`mlruns/` costs one entry instead of a walk. Empty outside a repository.
"""
listed = _git(
str(directory),
"ls-files",
"--others",
"--ignored",
"--exclude-standard",
"--directory",
"-z",
)
return frozenset(
directory / entry.rstrip("/") for entry in (listed or "").split("\0") if entry
)
def sync(
flows: Iterable[Flow],
client: Client,
+20
View File
@@ -376,3 +376,23 @@ def test_the_registry_and_the_builder_come_with_the_authoring_import():
assert fluksio.FLOWS is fluksio.sdk.FLOWS
assert fluksio.FLOWS["train"] is flow
assert fluksio.build(flow) == fluksio.sdk.build(flow)
def test_a_flow_records_the_module_that_declared_it():
"""What `fluksio run` imports by default, rather than the whole tree."""
from pathlib import Path
here = Path(__file__).resolve()
flow = a_flow()
assert flow.declared_in == str(here)
assert flow.module == __name__
origin = flow.document({"kind": "python", "repo": str(here.parents[2])})["origin"]
assert origin["declared_in"] == "tests/sdk/test_build.py"
assert origin["module"] == __name__
assert FlowDef.model_validate(flow.document(origin)).origin.module == __name__
# A repository the file is not in has no path to give it.
elsewhere = flow.document({"kind": "python", "repo": "/elsewhere"})["origin"]
assert (elsewhere["declared_in"], elsewhere["module"]) == ("", "")
+79
View File
@@ -550,6 +550,85 @@ def test_a_study_per_directory_imports_under_its_own_name(tmp_path) -> None:
sys.path[:] = [entry for entry in sys.path if entry != str(tmp_path / "dev")]
def test_run_imports_the_module_that_declared_the_flow(tmp_path, monkeypatch):
"""Not every script below the working directory: importing one runs it.
Imported under the name the last sync recorded, so the node bodies it
generates are the ones already stored. With nothing recorded, the walk.
"""
from fluksio.sdk import cli as sdk_cli
study = tmp_path / "dev" / "s1" / "study.py"
study.parent.mkdir(parents=True)
study.write_text("")
monkeypatch.chdir(tmp_path)
imported: list[tuple] = []
walked: list[list[str]] = []
monkeypatch.setattr(sdk_cli, "_import", lambda *args: imported.append(args))
monkeypatch.setattr(
sdk_cli,
"discover",
lambda targets, keep_going=False: walked.append(targets) or [],
)
class Engine:
origin: dict = {}
def get_flow(self, name):
return {"definition": {"origin": self.origin}}
engine = Engine()
engine.origin = {"declared_in": "dev/s1/study.py", "module": "s1.study"}
sdk_cli._sync_first(engine, [], "study")
assert imported == [
(str(tmp_path / "dev"), "s1.study", tmp_path / "dev" / "s1" / "study.py")
]
assert walked == []
# A name that does not match the file is not trusted, and neither is a
# file this checkout does not have: both walk, as an unrecorded flow does.
for origin in (
{"declared_in": "dev/s1/study.py", "module": "s2.study"},
{"declared_in": "dev/s9/study.py", "module": "s9.study"},
{},
):
engine.origin = origin
sdk_cli._sync_first(engine, [], "study")
assert walked == [["."], ["."], ["."]]
# And `--sync` is taken as said.
sdk_cli._sync_first(engine, ["dev/s1"], "study")
assert walked[-1] == ["dev/s1"]
def test_the_walk_skips_what_the_repository_ignores(tmp_path, monkeypatch) -> None:
"""`mlruns/` is output, not a study, and importing it would run it."""
import shutil
import subprocess
import sys
import pytest
from fluksio.sdk.cli import discover
if shutil.which("git") is None:
pytest.skip("needs git")
subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
(tmp_path / ".gitignore").write_text("mlruns/\nscratch.py\n")
(tmp_path / "mlruns" / "0").mkdir(parents=True)
(tmp_path / "mlruns" / "0" / "boom.py").write_text("raise SystemExit('imported')\n")
(tmp_path / "scratch.py").write_text("raise SystemExit('imported')\n")
(tmp_path / "walked_study.py").write_text("VALUE = 1\n")
try:
discover([str(tmp_path)])
assert sys.modules["walked_study"].VALUE == 1
finally:
sys.modules.pop("walked_study", None)
sys.path[:] = [entry for entry in sys.path if entry != str(tmp_path)]
def test_two_files_of_one_name_are_refused(tmp_path) -> None:
"""Python keeps one module per name, and a node's body imports by it.
+12 -7
View File
@@ -308,13 +308,18 @@ upgraded its cache is keyed on the whole repository, as it was before. See
fluksio run train --lr 0.05 --seed 7 [--wait]
```
Syncs the working directory *and everything under it*, then submits a run, so
the command after an edit is this one and nothing else, from the repository
root as readily as from the study's own folder. A study that will not import
is a warning rather than a stopped run; the upload is already a no-op for a
flow nothing changed in, so what the walk costs is importing the others.
`--sync dev/s1_baseline` (repeatable) narrows it to what you name when that
is not free, and `--no-sync` skips it entirely.
Syncs the flow, then submits a run, so the command after an edit is this one
and nothing else, from the repository root as readily as from the study's own
folder. What it imports is the module that declared the flow, as the last
`sync` recorded it, under the same module name. Importing a module runs it, so
the other scripts in the repository are left alone.
A flow the engine has no such record for (never synced, drawn on the canvas,
or on an engine older than the record) syncs the working directory *and
everything under it* instead, skipping whatever the repository's `.gitignore`
covers. A study there that will not import is a warning rather than a stopped
run. `--sync dev/s1_baseline` (repeatable) names what to sync outright, and
`--no-sync` skips it entirely.
Flags that are not its own are the flow's inputs, typed by what the flow
declares them as, so a name the flow has not got is refused by name, and
+10
View File
@@ -1159,6 +1159,16 @@ export const FlowOriginSchema = {
type: 'boolean',
title: 'Dirty',
default: false
},
declared_in: {
type: 'string',
title: 'Declared In',
default: ''
},
module: {
type: 'string',
title: 'Module',
default: ''
}
},
type: 'object',
+2
View File
@@ -364,6 +364,8 @@ export type FlowOrigin = {
repo?: string;
commit?: string;
dirty?: boolean;
declared_in?: string;
module?: string;
};
export type FlowRollup = {