Fleet

Run one command across every submodule of a superproject

Author

Benedict Thekkel

Discovery and reporting

Every verb walks the same list and reports the same way.

format_summary

def format_summary(
    rep, verb:str='ran'
):

The one line that replaces a message per submodule. Quiet submodules are counted, not narrated

Exported source

import re
import shutil
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from pathlib import Path

from fastcore.script import *

# The rendering layer lives in `automate`, the git/discovery layer in `github`. This
# module is the only one that knows about MANY repos; those two stay single-repo.
from nbdevAuto.automate import StepFailed, _push_summary, _run, _step, console, escape
from nbdevAuto.github import find_root, git, local_state, parse_gitmodules

Exported source

class Skip(Exception):
    "Raised by a walked callable that has decided this submodule has nothing to do"

@dataclass
class Outcome:
    "What happened in one submodule"
    path:   str
    state:  str                                  # ran | skipped | failed
    reason: str  = ""                            # why it was skipped, or how it failed
    detail: list = field(default_factory=list)   # short lines to show under the path

@dataclass
class Report:
    "Everything a `walk` did, and the exit code it implies"
    outcomes: list = field(default_factory=list)

    @property
    def ran(self):    return [o for o in self.outcomes if o.state == "ran"]
    @property
    def failed(self): return [o for o in self.outcomes if o.state == "failed"]
    @property
    def code(self):   return 1 if self.failed else 0

    def skip_counts(self):
        "Skip reason -> count, in the order the reasons were first seen"
        counts = {}
        for o in self.outcomes:
            if o.state == "skipped": counts[o.reason] = counts.get(o.reason, 0) + 1
        return counts

def format_summary(rep, verb="ran"):
    "The one line that replaces a message per submodule. Quiet submodules are counted, not narrated"
    n = len(rep.outcomes)
    parts = [f"{len(rep.ran)} {verb}"]
    parts += [f"{c} {reason}" for reason, c in rep.skip_counts().items()]
    if rep.failed: parts.append(f"{len(rep.failed)} FAILED")
    return f"{n} submodule{'' if n == 1 else 's'}: " + ", ".join(parts)

source

Report

def Report(
    outcomes:list=<factory>
)->None:

Everything a walk did, and the exit code it implies

source

Outcome

def Outcome(
    path:str, state:str, reason:str='', detail:list=<factory>
)->None:

What happened in one submodule

source

Skip

def Skip(
    *args, **kwargs
):

Raised by a walked callable that has decided this submodule has nothing to do

source

finish

def finish(
    rep, verb
):

Print the counted summary, name any failures, and return the exit code

Exported source

def _classify(base, sub, need_nbs, only_dirty):
    "The reason to skip this submodule before running anything, or None to run it"
    p = base/sub.path
    # An uninitialised submodule is an empty directory: present, but with no .git
    if not (p/".git").exists():             return "not initialised"
    if need_nbs and not (p/"nbs").is_dir(): return "without nbs"
    if only_dirty and not git(p, "status", "--porcelain"): return "clean"
    return None

def _fail_parts(e):
    """(headline, remaining lines) for a callable that did not finish.

    A `StepFailed` carries the command's own output, and its first line is the message
    worth putting on the summary line - "uncommitted changes" says far more than
    "update exited 1". The rest is kept for the block renderer."""
    if isinstance(e, StepFailed):
        out = (e.output or "").strip().splitlines()
        if out: return out[0], out[1:]
        return f"{e.cmd} exited {e.code}", []
    return f"{type(e).__name__}: {e}", []

def _render(o, style="block", width=0, header=True):
    "One submodule's result. A skip prints nothing at all; it is counted instead"
    if o.state == "skipped": return
    if style == "line":
        pad = f"{o.path:<{width}}"
        if o.state == "failed":
            console.print(f"[bold red]{escape(pad)}[/bold red] [red]{escape(o.reason)}[/red]")
            return
        console.print(f"[cyan]{escape(pad)}[/cyan] [dim]{escape(o.detail[0] if o.detail else '')}[/dim]")
        for line in o.detail[1:]: console.print(f"{' ' * width} [dim]{escape(line)}[/dim]")
        return
    if header: console.print(f"\n[bold cyan]==> {escape(o.path)}[/bold cyan]")
    if o.state == "failed":
        console.print(f"    [bold red]FAIL[/bold red] [red]{escape(o.reason)}[/red]")
    for line in o.detail: console.print(f"    [dim]{escape(line)}[/dim]")

def walk(fn, *, root=None, only_dirty=False, need_nbs=False, workers=1, style="block"):
    """Run `fn(path, sub)` in each submodule, returning a `Report`.

    `fn` returns a list of short strings to show under the submodule's name, or raises
    `Skip` once it knows there is nothing to do. Only submodules that do something print;
    the rest are counted by `format_summary`.

    `style="block"` gives each submodule a `==> path` heading with its own output beneath,
    which is what a long command like `upload` needs. `style="line"` gives one padded line
    per submodule, which suits verbs that report a single fact.

    Sequential block runs print the heading BEFORE calling `fn`, so its output streams
    underneath. Everything else prints after, in submodule order, so parallel work cannot
    interleave and two runs of the same fleet read the same way."""
    base = Path(root) if root else find_root()
    rep, todo = Report(), []
    for s in parse_gitmodules(base):
        why = _classify(base, s, need_nbs, only_dirty)
        if why: rep.outcomes.append(Outcome(s.path, "skipped", why))
        else:   todo.append(s)
    width = max((len(s.path) for s in todo), default=0) + 2

    def _one(s):
        try:                       return Outcome(s.path, "ran", detail=list(fn(base/s.path, s) or []))
        except Skip as e:          return Outcome(s.path, "skipped", str(e))
        except BaseException as e:
            head, rest = _fail_parts(e)
            return Outcome(s.path, "failed", head, rest)

    if workers > 1 and len(todo) > 1:
        with ThreadPoolExecutor(max_workers=workers) as ex: done = list(ex.map(_one, todo))
        for o in done: _render(o, style, width)
    else:
        done = []
        live = style == "block"
        for s in todo:
            if live: console.print(f"\n[bold cyan]==> {escape(s.path)}[/bold cyan]")
            o = _one(s)
            done.append(o)
            _render(o, style, width, header=not live)
    rep.outcomes += done
    rep.outcomes.sort(key=lambda o: o.path)
    return rep

def finish(rep, verb):
    "Print the counted summary, name any failures, and return the exit code"
    console.print(f"\n[bold]{escape(format_summary(rep, verb))}[/bold]")
    for o in rep.failed:
        console.print(f"[bold red]FAILED[/bold red] {escape(o.path)}: [dim]{escape(o.reason)}[/dim]")
    return rep.code

source

walk

def walk(
    fn, root:NoneType=None, only_dirty:bool=False, need_nbs:bool=False, workers:int=1, style:str='block'
):

Run fn(path, sub) in each submodule, returning a Report.

fn returns a list of short strings to show under the submodule’s name, or raises Skip once it knows there is nothing to do. Only submodules that do something print; the rest are counted by format_summary.

style="block" gives each submodule a ==> path heading with its own output beneath, which is what a long command like upload needs. style="line" gives one padded line per submodule, which suits verbs that report a single fact.

Sequential block runs print the heading BEFORE calling fn, so its output streams underneath. Everything else prints after, in submodule order, so parallel work cannot interleave and two runs of the same fleet read the same way.

Running a console script per submodule

source

upload_one

def upload_one(
    path, sub:NoneType=None, args:tuple=()
):

Run the single-repo upload in path, letting its own rich output through.

A subprocess rather than an in-process call: prep reads nbdev’s config from the cwd and caches it, and nbdev_test spawns its own process pool, so nineteen repos in one interpreter is a trap. The subprocess keeps this tty, so colour survives.

Exported source

def script_path(name):
    """The console script `name` from the environment this is running in.

    Taking it from `sys.executable`'s directory is what makes this work under a
    non-interactive shell that never put the venv on PATH."""
    cand = Path(sys.executable).parent/name
    if cand.is_file(): return str(cand)
    found = shutil.which(name)
    if found: return found
    raise RuntimeError(f"Cannot find the '{name}' console script. Run `uv sync` first")

def upload_one(path, sub=None, args=()):
    """Run the single-repo `upload` in `path`, letting its own rich output through.

    A subprocess rather than an in-process call: `prep` reads nbdev's config from the cwd
    and caches it, and `nbdev_test` spawns its own process pool, so nineteen repos in one
    interpreter is a trap. The subprocess keeps this tty, so colour survives."""
    r = subprocess.run([script_path("upload"), *args], cwd=str(path), check=False)
    if r.returncode: raise StepFailed("upload", r.returncode)
    return []

source

script_path

def script_path(
    name
):

The console script name from the environment this is running in.

Taking it from sys.executable’s directory is what makes this work under a non-interactive shell that never put the venv on PATH.

Verbs

source

fleet_upload

def fleet_upload(
    all:bool=False, # Upload every submodule with an nbs/ folder, not just the changed ones
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Run upload in each submodule that has local changes

Exported source

def fleet_upload(
    all:bool = False,    # Upload every submodule with an nbs/ folder, not just the changed ones
    root:str = None,     # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Run `upload` in each submodule that has local changes"
    # Sequential on purpose: `upload` runs nbdev_test with 8 workers, and knowledge-lab
    # has 4 vCPU. Parallel uploads would oversubscribe the box and interleave the output.
    rep = walk(upload_one, root=root, need_nbs=True, only_dirty=not all, workers=1)
    sys.exit(finish(rep, "uploaded"))

push and copy. Both report one line per submodule, and both take --dry-run.

source

copy_one

def copy_one(
    path, sub:NoneType=None, files:tuple=(), dest:str='nbs', dry_run:bool=False
):

Copy files into path/dest.

The file list belongs to the caller, so this stays a general-purpose copy rather than learning about any particular superproject’s shared chrome.

Exported source

def ahead_of_remote(path):
    "(commits origin has not seen, branch name). Branch is None when HEAD is detached"
    br = git(path, "rev-parse", "--abbrev-ref", "HEAD")
    if not br or br == "HEAD": return 0, None
    n = git(path, "rev-list", "--count", f"origin/{br}..HEAD")
    return (int(n) if n.isdigit() else 0), br

def push_one(path, sub=None, dry_run=False):
    "Push `path` when it has local commits origin has not seen"
    n, br = ahead_of_remote(path)
    if br is None:              raise Skip("detached HEAD")
    if not git(path, "remote"): raise Skip("no remote")
    if not n:                   raise Skip("nothing to push")
    if dry_run: return [f"would push {n} commit{'' if n == 1 else 's'} on {br}"]
    return _push_summary(_run("git", "-C", str(path), "push"))

def copy_one(path, sub=None, files=(), dest="nbs", dry_run=False):
    """Copy `files` into `path`/`dest`.

    The file list belongs to the caller, so this stays a general-purpose copy rather than
    learning about any particular superproject's shared chrome."""
    out = Path(path)/dest
    if not out.is_dir(): raise Skip(f"no {dest}/")
    names = []
    for f in files:
        src = Path(f)
        if not src.is_file(): raise StepFailed(f"copy {src.name}", 1, f"no such file: {src}")
        if not dry_run: shutil.copy2(src, out/src.name)
        names.append(src.name)
    return [("would copy " if dry_run else "copied ") + ", ".join(names) + f" -> {dest}/"]

source

push_one

def push_one(
    path, sub:NoneType=None, dry_run:bool=False
):

Push path when it has local commits origin has not seen

source

ahead_of_remote

def ahead_of_remote(
    path
):

(commits origin has not seen, branch name). Branch is None when HEAD is detached

source

fleet_copy

def fleet_copy(
    files:str, # Files to copy into each submodule, comma or space separated
    dest:str='nbs', # Directory inside each submodule to copy into
    dry_run:bool=False, # Report what would be copied without copying it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Copy shared files into every submodule that has the destination directory

Exported source

def fleet_push(
    dry_run:bool = False,   # Report what would be pushed without pushing it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Push every submodule that has local commits not yet on its remote"
    # Network-bound and independent per remote, so this one does parallelise.
    rep = walk(lambda p, s: push_one(p, s, dry_run=dry_run),
               root=root, workers=8, style="line")
    sys.exit(finish(rep, "would push" if dry_run else "pushed"))

def fleet_copy(
    files:str,              # Files to copy into each submodule, comma or space separated
    dest:str     = "nbs",   # Directory inside each submodule to copy into
    dry_run:bool = False,   # Report what would be copied without copying it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Copy shared files into every submodule that has the destination directory"
    fs = [f for f in re.split(r"[,\s]+", files) if f]
    if not fs: sys.exit("fleet copy: no files given")
    rep = walk(lambda p, s: copy_one(p, s, files=fs, dest=dest, dry_run=dry_run),
               root=root, workers=8, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

source

fleet_push

def fleet_push(
    dry_run:bool=False, # Report what would be pushed without pushing it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Push every submodule that has local commits not yet on its remote

status replaces both the git submodule status loop and the gss recipe.

source

fleet_status

def fleet_status(
    short:bool=False, # Only the submodules with changes, listing the changed paths
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Branch, dirty flag and HEAD for every submodule

Exported source

def _porcelain(path):
    """`git status -s` for `path`, NOT stripped.

    `github.git` strips its output, and `git status -s` pads the code to two columns, so
    stripping eats the leading space of the FIRST line and shifts that one path by a
    character. `automate.status` documents the same trap."""
    return subprocess.run(["git", "-C", str(path), "status", "-s"],
                          capture_output=True, text=True, check=False).stdout

def status_one(path, sub=None, short=False):
    """Branch, dirty flag and short HEAD for `path`.

    With `short`, clean submodules are skipped and the dirty ones list their changed
    paths - the fleet-wide "what have I got outstanding" view."""
    st = local_state(path)
    br = st["branch"] or "?"
    if br == "HEAD": br = "DETACHED"
    line = f"{br:<10} {'dirty' if st['dirty'] else 'clean':<6} {st['head'] or '-'}"
    if not short: return [line]
    if not st["dirty"]: raise Skip("clean")
    return [line] + [f"  {l}" for l in _porcelain(path).splitlines()]

def fleet_status(
    short:bool = False,   # Only the submodules with changes, listing the changed paths
    root:str   = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Branch, dirty flag and HEAD for every submodule"
    rep = walk(lambda p, s: status_one(p, s, short=short), root=root, workers=8, style="line")
    sys.exit(finish(rep, "changed" if short else "reported"))

source

status_one

def status_one(
    path, sub:NoneType=None, short:bool=False
):

Branch, dirty flag and short HEAD for path.

With short, clean submodules are skipped and the dirty ones list their changed paths - the fleet-wide “what have I got outstanding” view.

update and sync, the destructive pair. Fetching parallelises; checkout and merge do not.

source

pull_root

def pull_root(
    base
):

Fast-forward the superproject itself, with the advice its usual failure needs

Exported source

def _git_ok(path, *args):
    "True when git succeeded in `path`. `git()` hides the exit code, and here it matters"
    return subprocess.run(["git", "-C", str(path), *args],
                          capture_output=True, check=False).returncode == 0

def _count(path, rng):
    "Commit count for an `a..b` range, 0 when the range does not resolve"
    n = git(path, "rev-list", "--count", rng)
    return int(n) if n.isdigit() else 0

def fetch_all(root=None, workers=8):
    "Fetch every initialised submodule concurrently. This is the slow half of an update"
    base = Path(root) if root else find_root()
    subs = [s for s in parse_gitmodules(base) if (base/s.path/".git").exists()]
    with ThreadPoolExecutor(max_workers=workers) as ex:
        list(ex.map(lambda s: git(base/s.path, "fetch", "-q", "origin"), subs))
    return len(subs)

def update_one(path, sub=None, dry_run=False, branch="main"):
    """Reattach `path` to `branch` and fast-forward it to origin/<branch>.

    Assumes `fetch_all` already ran: fetching is the half that parallelises, and this is
    the half that must not, because a checkout rewrites the worktree."""
    if git(path, "status", "--porcelain"):
        raise StepFailed("update", 1, "uncommitted changes; commit or stash first")
    if not git(path, "remote"): raise Skip("no remote")
    target = f"origin/{branch}"
    if not git(path, "rev-parse", "--verify", "--quiet", target): raise Skip(f"no {target}")

    br   = git(path, "rev-parse", "--abbrev-ref", "HEAD")
    move = br != branch
    # Compare the branch we are ABOUT to be on, not the one we happen to be on now
    ref    = branch if _git_ok(path, "rev-parse", "--verify", "--quiet", branch) else "HEAD"
    behind = _count(path, f"{ref}..{target}")
    ahead  = _count(path, f"{target}..{ref}")
    # Diverged needs a human. The bash version echoed a line and moved on, which is easy
    # to scroll past; this fails the run so the exit code carries it.
    if ahead and behind:
        raise StepFailed("merge --ff-only", 1,
                         f"diverged from {target}: {ahead} ahead, {behind} behind - resolve by hand")
    was = "DETACHED" if br == "HEAD" else br
    if not move and not behind: raise Skip("up to date")
    if dry_run:
        bits = ([f"checkout {branch} (from {was})"] if move else []) + \
               ([f"fast-forward {behind}"] if behind else [])
        return ["would " + ", ".join(bits)]

    done = []
    if move:
        if not _git_ok(path, "checkout", "-q", branch):
            _run("git", "-C", str(path), "checkout", "-q", "-b", branch, target)
        done.append(f"checked out {branch} (was {was})")
    behind = _count(path, f"HEAD..{target}")   # the checkout may already have moved us
    if behind:
        _run("git", "-C", str(path), "merge", "--ff-only", target)
        done.append(f"fast-forwarded {behind} commit{'' if behind == 1 else 's'}")
    return done

def dirty_submodules(root=None):
    "Submodule paths holding uncommitted changes"
    base = Path(root) if root else find_root()
    return [s.path for s in parse_gitmodules(base)
            if (base/s.path/".git").exists() and git(base/s.path, "status", "--porcelain")]

def init_new(root=None, dry_run=False):
    """Initialise only the submodules git reports as uninitialised.

    A blanket `submodule update --init` would reset every initialised submodule back to
    its recorded gitlink and detach its HEAD, which is the one thing this setup avoids."""
    base = Path(root) if root else find_root()
    new = [l.split()[1] for l in git(base, "submodule", "status").splitlines() if l.startswith("-")]
    if new and not dry_run:
        _run("git", "-C", str(base), "submodule", "update", "--init", "--", *new)
    return new

def pull_root(base):
    "Fast-forward the superproject itself, with the advice its usual failure needs"
    try: return _run("git", "-C", str(base), "pull", "--ff-only").strip().splitlines()[-1:]
    except StepFailed as e:
        raise StepFailed(e.cmd, e.code, f"{e.output}\n"
            "If it complains about submodule paths, the recorded gitlinks drifted.\n"
            "Discard them (the update step re-advances them anyway):\n"
            "  git checkout -- $(git diff --name-only)") from None

source

init_new

def init_new(
    root:NoneType=None, dry_run:bool=False
):

Initialise only the submodules git reports as uninitialised.

A blanket submodule update --init would reset every initialised submodule back to its recorded gitlink and detach its HEAD, which is the one thing this setup avoids.

source

dirty_submodules

def dirty_submodules(
    root:NoneType=None
):

Submodule paths holding uncommitted changes

source

update_one

def update_one(
    path, sub:NoneType=None, dry_run:bool=False, branch:str='main'
):

Reattach path to branch and fast-forward it to origin/.

Assumes fetch_all already ran: fetching is the half that parallelises, and this is the half that must not, because a checkout rewrites the worktree.

source

fetch_all

def fetch_all(
    root:NoneType=None, workers:int=8
):

Fetch every initialised submodule concurrently. This is the slow half of an update

source

fleet_sync

def fleet_sync(
    dry_run:bool=False, # Report what would move without moving it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Bring this checkout level with every remote: pull, init new submodules, fast-forward each

Exported source

def fleet_update(
    dry_run:bool = False,   # Report what would move without moving it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Fetch every submodule, then fast-forward each one's main to origin/main"
    base = Path(root) if root else find_root()
    with _step(1, 2, "fetch every submodule") as d:
        d += [f"{fetch_all(base)} fetched"]
    console.print("[dim]2/2[/dim] [bold]fast-forward each main[/bold]")
    rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

def fleet_sync(
    dry_run:bool = False,   # Report what would move without moving it
    root:str     = None,    # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Bring this checkout level with every remote: pull, init new submodules, fast-forward each"
    base = Path(root) if root else find_root()
    # Pre-flight FIRST: a dirty submodule turns the checkout/merge below into a conflicted
    # tree, so nothing at all should happen until this passes.
    dirty = dirty_submodules(base)
    if dirty:
        console.print("[bold red]Uncommitted changes in:[/bold red]")
        for d in dirty: console.print(f"  {escape(d)}")
        console.print("Commit or stash them first ([bold]just upload[/bold]), then re-run.")
        sys.exit(1)

    with _step(1, 3, "pull the superproject") as d:
        d += ["skipped (dry run)"] if dry_run else pull_root(base)
    with _step(2, 3, "initialise new submodules") as d:
        new = init_new(base, dry_run=dry_run)
        d += [f"{'would init ' if dry_run else 'initialised '}{', '.join(new)}"] if new else ["none"]
    with _step(3, 3, "fetch every submodule") as d:
        d += [f"{fetch_all(base)} fetched"]
    rep = walk(lambda p, s: update_one(p, s, dry_run=dry_run), root=base, workers=1, style="line")
    sys.exit(finish(rep, "would update" if dry_run else "updated"))

source

fleet_update

def fleet_update(
    dry_run:bool=False, # Report what would move without moving it
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Fetch every submodule, then fast-forward each one’s main to origin/main

The `fleet` entry point

source

fleet

def fleet():

Dispatch fleet <verb> to the matching command.

upload and status are already console scripts bound to the single-repo versions in automate, so the fleet-wide ones cannot claim those names. One script with verbs sidesteps the collision and keeps --help working per verb.

Exported source

VERBS = {"upload": fleet_upload, "push": fleet_push, "copy": fleet_copy,
         "status": fleet_status, "update": fleet_update, "sync": fleet_sync}

def parse_verb(argv):
    "(verb, remaining args) for `fleet ...`, or None when `argv` does not name a verb"
    if not argv or argv[0].startswith("-"): return None
    return (argv[0], argv[1:]) if argv[0] in VERBS else None

def usage():
    "The verb list, for --help and for an unknown verb"
    lines = ["Usage: fleet <verb> [options]", "", "Verbs:"]
    lines += [f"  {name:<8} {(fn.__doc__ or '').strip()}" for name, fn in sorted(VERBS.items())]
    lines += ["", "Any verb takes --help for its own options."]
    return "\n".join(lines)

def run_verb(name, argv):
    """Parse `argv` against the verb's own annotations, then call it.

    Deliberately NOT `@call_parse`: fastcore runs the first decorated function in a module
    the moment that module is `__main__`, which is exactly what makes `python -m
    nbdevAuto.github` work with its single verb. With several verbs it would run whichever
    happened to be defined first, so the verbs stay plain functions and the parser is built
    here. `anno_parser` is the same one `call_parse` uses, so `--help` is unchanged."""
    fn = VERBS[name]
    args = vars(anno_parser(fn, prog=f"fleet {name}").parse_args(argv))
    for k in ("pdb", "xtra"): args.pop(k, None)
    return fn(**args)

def fleet():
    """Dispatch `fleet <verb>` to the matching command.

    `upload` and `status` are already console scripts bound to the single-repo versions in
    `automate`, so the fleet-wide ones cannot claim those names. One script with verbs
    sidesteps the collision and keeps `--help` working per verb."""
    picked = parse_verb(sys.argv[1:])
    if not picked:
        print(usage())
        sys.exit(0 if sys.argv[1:2] in ([], ["-h"], ["--help"]) else 2)
    return run_verb(*picked)

source

run_verb

def run_verb(
    name, argv
):

Parse argv against the verb’s own annotations, then call it.

Deliberately NOT @call_parse: fastcore runs the first decorated function in a module the moment that module is __main__, which is exactly what makes python -m nbdevAuto.github work with its single verb. With several verbs it would run whichever happened to be defined first, so the verbs stay plain functions and the parser is built here. anno_parser is the same one call_parse uses, so --help is unchanged.

source

usage

def usage():

The verb list, for –help and for an unknown verb

source

parse_verb

def parse_verb(
    argv
):

(verb, remaining args) for fleet ..., or None when argv does not name a verb

python -m nbdevAuto.fleet is the other way in, and the one the justfile falls back to when the installed nbdevAuto predates this module. #|export puts the guard in the library; #|eval: false keeps nbdev_test from running the CLI while testing the notebook.

Tests

A throwaway superproject of real git repos, rebuilt per run.

from fastcore.test import *
import tempfile

def _fixture(specs=(("a_dirty", True, True), ("b_clean", True, False), ("c_nonbs", False, False))):
    "A throwaway superproject: (name, has nbs/, is dirty) per submodule"
    d = Path(tempfile.mkdtemp())
    lines = []
    for name, nbs, dirty in specs:
        p = d/name; p.mkdir()
        subprocess.run(["git", "init", "-q", "-b", "main", str(p)], check=True)
        for k, v in (("user.email", "t@example.com"), ("user.name", "t")):
            subprocess.run(["git", "-C", str(p), "config", k, v], check=True)
        (p/"README.md").write_text("x")
        if nbs: (p/"nbs").mkdir()
        subprocess.run(["git", "-C", str(p), "add", "."], check=True)
        subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", "init"], check=True)
        if dirty: (p/"README.md").write_text("changed")
        lines.append(f'[submodule "{name}"]\n\tpath = {name}\n\turl = git@github.com:me/{name}.git\n')
    (d/".gitmodules").write_text("".join(lines))
    return d

def _with_origin(p, extra=1):
    "Give the repo at `p` a real bare origin, then leave it `extra` commits ahead of it"
    bare = Path(f"{p}.origin.git")
    subprocess.run(["git", "init", "-q", "--bare", str(bare)], check=True)
    subprocess.run(["git", "-C", str(p), "remote", "add", "origin", str(bare)], check=True)
    subprocess.run(["git", "-C", str(p), "push", "-q", "-u", "origin", "main"], check=True)
    for i in range(extra):
        (p/f"extra{i}.txt").write_text("x")
        subprocess.run(["git", "-C", str(p), "add", f"extra{i}.txt"], check=True)
        subprocess.run(["git", "-C", str(p), "commit", "-q", "-m", f"extra{i}"], check=True)
    return bare

_root = _fixture()

# Only the dirty submodule with an nbs/ folder runs; the other two are counted, not narrated.
seen = []
rep = walk(lambda p, s: seen.append(s.path), root=_root, need_nbs=True, only_dirty=True, style="line")
test_eq(seen, ["a_dirty"])
test_eq(len(rep.ran), 1)
test_eq(rep.skip_counts(), {"without nbs": 1, "clean": 1})
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 clean, 1 without nbs")
test_eq(rep.code, 0)

# Dropping the dirty filter picks up both nbs/ submodules.
rep = walk(lambda p, s: None, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.ran], ["a_dirty", "b_clean"])
test_eq(format_summary(rep, "uploaded"), "3 submodules: 2 uploaded, 1 without nbs")

# A failure is named and counted, does not abort the others, and sets the exit code.
def _boom(p, s):
    if s.path == "a_dirty": raise StepFailed("upload", 2, "fatal: nope")
    return ["fine"]
rep = walk(_boom, root=_root, need_nbs=True, style="line")
test_eq([o.path for o in rep.failed], ["a_dirty"])
# A StepFailed's own message is the headline, because "upload exited 2" says nothing
test_eq(rep.failed[0].reason, "fatal: nope")
test_eq([o.path for o in rep.ran], ["b_clean"])
test_eq(rep.code, 1)
test_eq(format_summary(rep, "uploaded"), "3 submodules: 1 uploaded, 1 without nbs, 1 FAILED")

# With no output to quote, the command and exit code are the fallback headline.
def _silent(p, s): raise StepFailed("upload", 2)
test_eq(walk(_silent, root=_root, need_nbs=True, style="line").failed[0].reason, "upload exited 2")

# An uninitialised submodule is a directory with no .git, and is reported rather than run.
_bare = _fixture()
import shutil as _sh; _sh.rmtree(_bare/"b_clean"/".git")
test_eq(walk(lambda p, s: None, root=_bare, style="line").skip_counts(), {"not initialised": 1})

# Parallel and sequential runs agree, and both report in submodule order.
rep_p = walk(lambda p, s: [s.path], root=_root, workers=4, style="line")
rep_s = walk(lambda p, s: [s.path], root=_root, workers=1, style="line")
test_eq([o.path for o in rep_p.outcomes], [o.path for o in rep_s.outcomes])
test_eq([o.detail for o in rep_p.outcomes], [o.detail for o in rep_s.outcomes])

a_dirty   
a_dirty   
b_clean   
a_dirty   fatal: nope
b_clean   fine
a_dirty   upload exited 2
b_clean   upload exited 2
a_dirty   
c_nonbs   
a_dirty   a_dirty
b_clean   b_clean
c_nonbs   c_nonbs
a_dirty   a_dirty
b_clean   b_clean
c_nonbs   c_nonbs

# Verb dispatch: a verb, its args, and everything that is not one.
test_eq(parse_verb(["upload", "--all"]), ("upload", ["--all"]))
test_eq(parse_verb(["upload"]), ("upload", []))
test_eq(parse_verb([]), None)
test_eq(parse_verb(["--help"]), None)
test_eq(parse_verb(["nosuchverb"]), None)
assert "upload" in usage() and "Usage: fleet" in usage()

# `run_verb` parses against the verb's own signature. Checked on a throwaway verb so the
# test does not have to run a real upload.
def _echo(all:bool=False, root:str=None):
    "Echo verb, for tests"
    return (all, root)
VERBS["_echo"] = _echo
try:
    test_eq(run_verb("_echo", []), (False, None))
    test_eq(run_verb("_echo", ["--all"]), (True, None))
    test_eq(run_verb("_echo", ["--root", "/tmp/x"]), (False, "/tmp/x"))
    test_eq(parse_verb(["_echo", "--all"]), ("_echo", ["--all"]))
finally:
    del VERBS["_echo"]

# `script_path` says what to do rather than failing obscurely.
try:
    script_path("definitely-not-a-console-script-xyz")
    assert False, "a missing script must raise"
except RuntimeError as e:
    assert "uv sync" in str(e)

Phase 2: push and copy

# --- push -------------------------------------------------------------------------
_r2 = _fixture()
_with_origin(_r2/"b_clean", extra=2)
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
test_eq(push_one(_r2/"b_clean", dry_run=True), ["would push 2 commits on main"])
# dry_run must not move the remote
test_eq(ahead_of_remote(_r2/"b_clean"), (2, "main"))
push_one(_r2/"b_clean")
test_eq(ahead_of_remote(_r2/"b_clean"), (0, "main"))

# The three reasons a submodule is passed over, each named rather than silently ignored.
def _skip_reason(fn, *a, **kw):
    try: fn(*a, **kw); return None
    except Skip as e: return str(e)
test_eq(_skip_reason(push_one, _r2/"b_clean"), "nothing to push")
test_eq(_skip_reason(push_one, _r2/"a_dirty"), "no remote")
subprocess.run(["git", "-C", str(_r2/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
test_eq(_skip_reason(push_one, _r2/"c_nonbs"), "detached HEAD")

# A runtime Skip is counted, not printed, and never a failure.
rep = walk(lambda p, s: push_one(p, s, dry_run=True), root=_r2, style="line")
test_eq(rep.ran, [])
test_eq(rep.skip_counts(), {"no remote": 1, "nothing to push": 1, "detached HEAD": 1})
test_eq(rep.code, 0)

# --- copy -------------------------------------------------------------------------
_r3 = _fixture()
_shared = _r3/"shared.yml"; _shared.write_text("a: 1")
test_eq(copy_one(_r3/"a_dirty", files=[_shared], dry_run=True), ["would copy shared.yml -> nbs/"])
assert not (_r3/"a_dirty"/"nbs"/"shared.yml").exists(), "dry_run wrote a file"
test_eq(copy_one(_r3/"a_dirty", files=[_shared]), ["copied shared.yml -> nbs/"])
test_eq((_r3/"a_dirty"/"nbs"/"shared.yml").read_text(), "a: 1")

# No destination directory is a skip; a missing SOURCE is a failure, not a quiet no-op.
test_eq(_skip_reason(copy_one, _r3/"c_nonbs", files=[_shared]), "no nbs/")
try:
    copy_one(_r3/"a_dirty", files=["/nope/missing.yml"])
    assert False, "a missing source must fail"
except StepFailed as e:
    assert "no such file" in e.output

rep = walk(lambda p, s: copy_one(p, s, files=[_shared]), root=_r3, workers=4, style="line")
test_eq(len(rep.ran), 2)
test_eq(format_summary(rep, "updated"), "3 submodules: 2 updated, 1 no nbs/")

Phases 4 and 5: status, update, sync

# --- status -----------------------------------------------------------------------
_r4 = _fixture()
_line = status_one(_r4/"a_dirty")[0]
assert _line.startswith("main") and "dirty" in _line
assert "clean" in status_one(_r4/"b_clean")[0]
# --short lists the changed paths, and passes over the clean ones
_out = status_one(_r4/"a_dirty", short=True)
assert any("README.md" in l for l in _out[1:]), _out
# The FIRST porcelain line must not be shifted: `git status -s` pads the code to two
# columns and a strip would eat that leading space on line one only.
test_eq([l[:5] for l in _out[1:]], ["   M "] * len(_out[1:]))
test_eq(_skip_reason(status_one, _r4/"b_clean", short=True), "clean")
rep = walk(lambda p, s: status_one(p, s, short=True), root=_r4, style="line")
test_eq(format_summary(rep, "changed"), "3 submodules: 1 changed, 2 clean")

# A detached HEAD is named, not shown as a branch called HEAD
subprocess.run(["git", "-C", str(_r4/"c_nonbs"), "checkout", "-q", "--detach"], check=True)
assert status_one(_r4/"c_nonbs")[0].startswith("DETACHED")

# --- update -----------------------------------------------------------------------
_r5 = _fixture()
_p = _r5/"b_clean"
_with_origin(_p, extra=2)
subprocess.run(["git", "-C", str(_p), "push", "-q"], check=True)          # remote now level
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~2"], check=True)
test_eq(_count(_p, "HEAD..origin/main"), 2)                                # 2 behind

test_eq(update_one(_p, dry_run=True), ["would fast-forward 2"])
test_eq(_count(_p, "HEAD..origin/main"), 2)                                # dry run moved nothing
test_eq(update_one(_p), ["fast-forwarded 2 commits"])
test_eq(_count(_p, "HEAD..origin/main"), 0)
test_eq(_skip_reason(update_one, _p), "up to date")

# Diverged is a FAILURE, not a quiet skip: it needs a human and the exit code must say so.
subprocess.run(["git", "-C", str(_p), "reset", "-q", "--hard", "HEAD~1"], check=True)
(_p/"mine.txt").write_text("mine")
subprocess.run(["git", "-C", str(_p), "add", "mine.txt"], check=True)
subprocess.run(["git", "-C", str(_p), "commit", "-q", "-m", "mine"], check=True)
try:
    update_one(_p)
    assert False, "a diverged submodule must fail"
except StepFailed as e:
    assert "diverged" in e.output and "resolve by hand" in e.output

# A dirty worktree is refused before anything is checked out.
try:
    update_one(_r5/"a_dirty")
    assert False, "a dirty submodule must fail"
except StepFailed as e:
    assert "uncommitted changes" in e.output
test_eq(_skip_reason(update_one, _r5/"c_nonbs"), "no remote")

# A detached HEAD is reattached to main.
_q = _fixture()/"a_dirty"
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--", "README.md"], check=True)
_with_origin(_q, extra=0)
subprocess.run(["git", "-C", str(_q), "checkout", "-q", "--detach"], check=True)
test_eq(update_one(_q, dry_run=True), ["would checkout main (from DETACHED)"])
test_eq(update_one(_q), ["checked out main (was DETACHED)"])
test_eq(git(_q, "rev-parse", "--abbrev-ref", "HEAD"), "main")

# --- sync helpers -----------------------------------------------------------------
test_eq(dirty_submodules(_r4), ["a_dirty"])
# init_new only ever touches submodules git itself calls uninitialised
test_eq(init_new(_r4, dry_run=True), [])

Discovery and reporting

format_summary

Report

Outcome

Skip

finish

walk

Running a console script per submodule

upload_one

script_path

Verbs

fleet_upload

copy_one

push_one

ahead_of_remote

fleet_copy

fleet_push

fleet_status

status_one

pull_root

init_new

dirty_submodules

update_one

fetch_all

fleet_sync

fleet_update

The fleet entry point

fleet

run_verb

usage

parse_verb

Tests

Phase 2: push and copy

Phases 4 and 5: status, update, sync

The `fleet` entry point