diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index b877953..40f04c1 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -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): diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index 5e0cf6b..888edf8 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -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]: diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index 242b491..7393d56 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -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") diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 7c57bfb..77cd89c 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -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, diff --git a/backend/tests/sdk/test_build.py b/backend/tests/sdk/test_build.py index 2f68b3d..672bc5b 100644 --- a/backend/tests/sdk/test_build.py +++ b/backend/tests/sdk/test_build.py @@ -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"]) == ("", "") diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index cb74c49..9202efa 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -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. diff --git a/docs/code/cli.md b/docs/code/cli.md index 488096a..70ae500 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -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 diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index ec3cbfa..f846aa5 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -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', diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 4095aee..a10b029 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -364,6 +364,8 @@ export type FlowOrigin = { repo?: string; commit?: string; dirty?: boolean; + declared_in?: string; + module?: string; }; export type FlowRollup = {