GitHub

Remote CI, Pages and branch status for every submodule, over the GitHub API
Author

Benedict Thekkel

The justfile recipes in the Knowledge superproject are all local git: fetch, checkout, merge --ff-only, push. Nothing in them can see what happened after a push, so a repo whose nbdev3-ci run failed (the classic “Notebooks and library are not in sync”) or whose Pages deploy never landed looks identical to a healthy one from the terminal.

This module closes that gap. It reads .gitmodules, then asks the GitHub API, concurrently, for four things per repo: the newest CI run, the newest Pages deploy run, the latest Pages build, and the remote main sha. githubkit is an optional dependency (pip install 'nbdevAuto[gh]'), so every import of it is deferred to the function that needs it.

Exported source 
import asyncio
import os
import re
import subprocess
from collections import namedtuple
from pathlib import Path

from fastcore.script import *

Submodule = namedtuple("Submodule", "path owner repo")

Repo discovery

The list of repos comes from .gitmodules rather than a hardcoded table, so a newly added submodule is covered the moment it is registered. Entries are keyed by path, not by the section name: several names in this repo have drifted from the path they point at ([submodule "DL/Research"] has path = DL/DL_tasks).

source

find_root

def find_root(
    start:NoneType=None
):

Nearest ancestor of start (default: the cwd) holding a .gitmodules file

Exported source 
def find_root(start=None):
    "Nearest ancestor of `start` (default: the cwd) holding a .gitmodules file"
    p = Path(start or Path.cwd()).resolve()
    for d in (p, *p.parents):
        if (d/".gitmodules").is_file(): return d
    raise FileNotFoundError("No .gitmodules in this directory or any parent")

source

parse_gitmodules

def parse_gitmodules(
    root:NoneType=None
):

Submodule(path, owner, repo) for every entry in /.gitmodules, sorted by path

Exported source 
def _gitmodules_kv(root):
    "({section name: path}, {section name: url}) from <root>/.gitmodules"
    cfg = Path(root)/".gitmodules"
    # An explicit --root that holds no .gitmodules must fail, not report an empty fleet
    if not cfg.is_file(): raise FileNotFoundError(f"No .gitmodules at {cfg}")
    out = subprocess.run(["git", "config", "-f", str(cfg),
                          "--get-regexp", r"^submodule\..*\.(path|url)$"],
                         capture_output=True, text=True, check=False).stdout
    paths, urls = {}, {}
    for line in out.splitlines():
        key, _, val = line.partition(" ")
        name, _, field = key[len("submodule."):].rpartition(".")
        (paths if field == "path" else urls)[name] = val
    return paths, urls

def parse_gitmodules(root=None):
    "Submodule(path, owner, repo) for every entry in <root>/.gitmodules, sorted by path"
    root = Path(root) if root else find_root()
    paths, urls = _gitmodules_kv(root)
    subs = []
    for name, path in paths.items():
        # git@github.com:owner/repo.git and https://github.com/owner/repo both land here
        m = re.search(r"[:/]([^/:]+)/([^/]+?)(?:\.git)?$", urls.get(name, ""))
        if m: subs.append(Submodule(path, m[1], m[2]))
    return sorted(subs, key=lambda s: s.path)
from fastcore.test import *
import tempfile

FIXTURE = """[submodule "DL/Research"]
    path = DL/DL_tasks
    url = git@github.com:bthek1/DL_tasks.git
[submodule "Personal/bthek1"]
    path = Personal/bthek1
    url = https://github.com/bthek1/bthek1
"""

with tempfile.TemporaryDirectory() as d:
    (Path(d)/".gitmodules").write_text(FIXTURE)
    test_eq(find_root(d), Path(d).resolve())
    test_eq(parse_gitmodules(d), [Submodule("DL/DL_tasks", "bthek1", "DL_tasks"),
                                  Submodule("Personal/bthek1", "bthek1", "bthek1")])

with tempfile.TemporaryDirectory() as d:
    test_fail(lambda: find_root(d), contains="No .gitmodules")
    test_fail(lambda: parse_gitmodules(d), contains="No .gitmodules")   # never an empty fleet

Auth

No new secret: the token comes from the gh login that is already on every machine here. An explicit $GITHUB_TOKEN / $GH_TOKEN wins, which is what CI would set.

source

resolve_token

def resolve_token():

GitHub token from \(GITHUB_TOKEN/\)GH_TOKEN, else gh auth token

Exported source 
def resolve_token():
    "GitHub token from $GITHUB_TOKEN/$GH_TOKEN, else `gh auth token`"
    for var in ("GITHUB_TOKEN", "GH_TOKEN"):
        if os.environ.get(var): return os.environ[var]
    try:
        r = subprocess.run(["gh", "auth", "token"], capture_output=True, text=True,
                           timeout=10, check=False)
        if r.returncode == 0 and r.stdout.strip(): return r.stdout.strip()
    except (FileNotFoundError, subprocess.TimeoutExpired): pass
    raise RuntimeError("No GitHub token: run `gh auth login`, or set GITHUB_TOKEN")

Local state

The local sha stays a local lookup. Comparing it against the remote sha is what tells you a submodule has unpushed work, and it costs nothing.

source

local_sha

def local_sha(
    path, n:int=7
):

Short HEAD sha of the checkout at path, or None when it is not one

Exported source 
def local_sha(path, n=7):
    "Short HEAD sha of the checkout at `path`, or None when it is not one"
    r = subprocess.run(["git", "-C", str(path), "rev-parse", f"--short={n}", "HEAD"],
                       capture_output=True, text=True, check=False)
    return r.stdout.strip() or None

Remote state

Three independent lookups per repo, each returning None rather than raising when the thing does not exist. That is the normal case, not an error: Personal/bthek1 is the GitHub profile repo, so it has no test.yaml, no deploy.yaml and no Pages site, and it must render as three dashes.

Workflows are addressed by filename with a .yaml/.yml fallback, since that is what fastai/workflows scaffolds.

Only a 404 counts as “not there”. A 401 or a 403 means the token is bad or rate limited, and swallowing those would paint the whole fleet with dashes and exit 0, which reads as healthy. Those propagate.

source

api_error

def api_error(
    e
):

Readable one-liner for a githubkit failure

Exported source 
def _missing(e):
    "True when a githubkit failure means the thing does not exist, rather than that we cannot ask"
    return getattr(getattr(e, "response", None), "status_code", None) == 404

def api_error(e):
    "Readable one-liner for a githubkit failure"
    code = getattr(getattr(e, "response", None), "status_code", None)
    if code == 401: return "GitHub rejected the token (401): run `gh auth login`, or refresh $GITHUB_TOKEN"
    if code == 403: return "GitHub refused the request (403): rate limited, or the token lacks scope"
    return f"GitHub API error: {e}"

source

newest_run

def newest_run(
    runs
):

The newest run in runs, by run_number, which is monotonic per workflow.

Do NOT just take runs[0]. This endpoint is eventually consistent and its ordering is not a contract: on 2026-08-26 a per_page=1 query for ML_methods returned run number 96 (2026-02-17, failure) when 107 (2026-08-16, success) was the newest, and just report called a healthy repo FAILURE. A single-item page gives the caller no way to notice. Asking for several and taking the maximum costs the same one request.

Exported source 
CI_WORKFLOWS    = ("test.yaml", "test.yml")       # fastai/workflows/nbdev3-ci
PAGES_WORKFLOWS = ("deploy.yaml", "deploy.yml")   # fastai/workflows/quarto-ghp3

def newest_run(runs):
    """The newest run in `runs`, by `run_number`, which is monotonic per workflow.

    Do NOT just take `runs[0]`. This endpoint is eventually consistent and its ordering is
    not a contract: on 2026-08-26 a `per_page=1` query for ML_methods returned run number
    96 (2026-02-17, failure) when 107 (2026-08-16, success) was the newest, and `just
    report` called a healthy repo FAILURE. A single-item page gives the caller no way to
    notice. Asking for several and taking the maximum costs the same one request."""
    numbered = [r for r in runs if getattr(r, "run_number", None) is not None]
    return max(numbered, key=lambda r: r.run_number) if numbered else runs[0]

async def _latest_run(gh, sub, candidates, branch="main"):
    "Newest run of the first workflow in `candidates` that exists, or None"
    from githubkit.exception import RequestFailed
    for wf in candidates:
        try:
            r = await gh.rest.actions.async_list_workflow_runs(
                sub.owner, sub.repo, wf, branch=branch, per_page=10)
        except RequestFailed as e:
            if _missing(e): continue
            raise
        runs = r.parsed_data.workflow_runs
        if runs:
            run = newest_run(runs)
            # conclusion is None while a run is still going, status covers that case
            return {"state": run.conclusion or run.status, "sha": run.head_sha, "url": run.html_url}
    return None
# `newest_run` exists because the API's ordering is not trustworthy: a per_page=1 query
# once returned a six-month-old failure as "newest" and reported a healthy repo as FAILURE.
from types import SimpleNamespace as _NS
_runs = [_NS(run_number=96, conclusion="failure"), _NS(run_number=107, conclusion="success"),
         _NS(run_number=103, conclusion="failure")]
test_eq(newest_run(_runs).run_number, 107)          # order in the page is irrelevant
test_eq(newest_run(_runs[::-1]).run_number, 107)
test_eq(newest_run([_runs[0]]).run_number, 96)      # a single run is still the newest one
# No run_number at all (should not happen) falls back to the API's own first item
_bare = [_NS(run_number=None, conclusion="success"), _NS(run_number=None, conclusion="failure")]
test_eq(newest_run(_bare).conclusion, "success")

source

gather_status

async def gather_status(
    subs, token:NoneType=None, root:NoneType=None
):

Every row, concurrently, through one client and one connection pool

Exported source 
async def _pages_build(gh, sub):
    "Status of the latest Pages build, or None when Pages is not enabled"
    from githubkit.exception import RequestFailed
    try: r = await gh.rest.repos.async_get_latest_pages_build(sub.owner, sub.repo)
    except RequestFailed as e:
        if _missing(e): return None
        raise
    return r.parsed_data.status

async def _remote_sha(gh, sub, branch="main", n=7):
    "Short sha that `branch` points at on the remote, or None"
    from githubkit.exception import RequestFailed
    try: r = await gh.rest.git.async_get_ref(sub.owner, sub.repo, f"heads/{branch}")
    except RequestFailed as e:
        if _missing(e): return None
        raise
    return r.parsed_data.object_.sha[:n]
Exported source 
async def repo_status(gh, sub, root=None):
    "One row: CI run, Pages deploy run, Pages build, and local vs remote sha"
    ci, deploy, built, remote = await asyncio.gather(
        _latest_run(gh, sub, CI_WORKFLOWS), _latest_run(gh, sub, PAGES_WORKFLOWS),
        _pages_build(gh, sub), _remote_sha(gh, sub))
    return {"path": sub.path, "owner": sub.owner, "repo": sub.repo,
            "ci": ci, "deploy": deploy, "built": built, "remote": remote,
            "local": local_sha(Path(root or find_root())/sub.path)}

def client(token=None):
    "One githubkit client, with githubkit's response cache turned off"
    try:
        from githubkit import GitHub
    except ImportError:
        raise ImportError("githubkit is not installed: pip install 'nbdevAuto[gh]'") from None
    # githubkit defaults to http_cache=True. Its default MemCacheStrategy sounds
    # like a dict but is not: githubkit's AsyncMemoryStorage subclasses hishel's
    # AsyncSqliteStorage over a ":memory:" database, so every cached response is a
    # transaction on one shared anysqlite connection. This tool issues ~76 requests
    # concurrently through a single client, and those transactions collide: runs
    # die intermittently on "cannot start a transaction within a transaction" or
    # "bad parameter or other API misuse" and exit 3, which reads as "cannot ask
    # GitHub" on a perfectly healthy fleet. Observed about twice in 40 runs.
    #
    # Caching is the wrong default for this question anyway - the whole point is to
    # see whether the thing just pushed has built yet - and 76 calls against a
    # 5000/hour budget do not need it. Turning it off removes the shared connection
    # from the concurrent path entirely.
    return GitHub(token or resolve_token(), http_cache=False)

async def gather_status(subs, token=None, root=None):
    "Every row, concurrently, through one client and one connection pool"
    async with client(token) as gh:
        return await asyncio.gather(*[repo_status(gh, s, root) for s in subs])

source

client

def client(
    token:NoneType=None
):

One githubkit client, with githubkit’s response cache turned off

source

repo_status

async def repo_status(
    gh, sub, root:NoneType=None
):

One row: CI run, Pages deploy run, Pages build, and local vs remote sha

Rendering

Same visual language as just report: bold header, and colour only where something wants attention. A healthy fleet prints plain white.

source

format_table

def format_table(
    rows, color:bool=True
):

The status table as a string, one line per submodule

Exported source 
FAILED  = {"failure", "timed_out", "cancelled", "action_required", "startup_failure"}
RUNNING = {"in_progress", "queued", "requested", "waiting", "pending"}
BOLD, RED, YELLOW, CYAN, DIM, OFF = "\033[1m", "\033[31m", "\033[33m", "\033[36m", "\033[2m", "\033[0m"

def _state(v):
    "Display text and colour for one conclusion / build status"
    if v is None:     return "-", DIM
    if v in FAILED:   return v.upper(), RED
    if v in RUNNING:  return v, YELLOW
    return v, ""

def format_table(rows, color=True):
    "The status table as a string, one line per submodule"
    def c(s, col): return f"{col}{s}{OFF}" if color and col else s
    head = f"{'SUBMODULE':<30} {'CI':<10} {'PAGES':<10} {'BUILT':<9} {'LOCAL':<8} {'REMOTE':<8}"
    out = [c(head, BOLD)]
    for r in rows:
        ci, ci_c = _state(r["ci"]["state"] if r["ci"] else None)
        dp, dp_c = _state(r["deploy"]["state"] if r["deploy"] else None)
        bl, bl_c = _state(r["built"])
        # a local/remote mismatch is unpushed or unpulled work, not a failure
        sync = CYAN if (r["local"] and r["remote"] and r["local"] != r["remote"]) else ""
        out.append(" ".join([f"{r['path']:<30}", c(f"{ci:<10}", ci_c), c(f"{dp:<10}", dp_c),
                             c(f"{bl:<9}", bl_c), c(f"{r['local'] or '-':<8}", sync),
                             c(f"{r['remote'] or '-':<8}", sync)]))
    return "\n".join(out)

source

has_failure

def has_failure(
    rows
):

True when any workflow failed or a Pages build errored

Exported source 
def failures(rows):
    "(path, kind, url) for every workflow run that did not succeed"
    return [(r["path"], kind, r[kind]["url"])
            for r in rows for kind in ("ci", "deploy")
            if r[kind] and r[kind]["state"] in FAILED]

def has_failure(rows):
    "True when any workflow failed or a Pages build errored"
    return bool(failures(rows)) or any(r["built"] not in (None, "built") for r in rows)

source

failures

def failures(
    rows
):

(path, kind, url) for every workflow run that did not succeed

ok   = dict(path="Web/WEB_doc", owner="bthek1", repo="WEB_doc", built="built",
            ci={"state":"success","sha":"a"*40,"url":"https://x/1"},
            deploy={"state":"success","sha":"a"*40,"url":"https://x/2"},
            local="abc1234", remote="abc1234")
bare = dict(path="Personal/bthek1", owner="bthek1", repo="bthek1", ci=None, deploy=None,
            built=None, local="0ea0091", remote="0ea0091")
bad  = {**ok, "path":"DL/DL_tasks", "ci":{"state":"failure","sha":"b"*40,"url":"https://x/3"}}
ahead = {**ok, "path":"ML/ML_methods", "local":"1111111", "remote":"2222222"}

lines = format_table([ok, bare, bad, ahead], color=False).splitlines()
test_eq(len(lines), 5)
test_eq(lines[2].split(), ["Personal/bthek1", "-", "-", "-", "0ea0091", "0ea0091"])  # no workflows, no Pages
test_eq(lines[3].split()[1], "FAILURE")
test_eq(failures([ok, bare, bad, ahead]), [("DL/DL_tasks", "ci", "https://x/3")])
test_eq(has_failure([ok, bare, ahead]), False)
test_eq(has_failure([ok, bad]), True)
test_eq(has_failure([{**ok, "built":"errored"}]), True)

class _Resp:
    def __init__(self, code): self.status_code = code
class _Failed(Exception):
    def __init__(self, code): self.response = _Resp(code)

test_eq(_missing(_Failed(404)), True)      # repo has no Pages / no such workflow: a normal row
test_eq(_missing(_Failed(401)), False)     # bad token: must propagate, not render as dashes
test_eq(_missing(_Failed(403)), False)
assert "401" in api_error(_Failed(401)) and "gh auth login" in api_error(_Failed(401))
assert "403" in api_error(_Failed(403))

Local state, without the fetch storm

just report used to run git fetch in all 19 submodules just to count how far each one is behind. That is the slowest thing in the justfile, and nearly all of it is wasted: nothing moved in most of those repos.

The API already said what sha main points at on the remote, so a fetch is only needed when this checkout does not already have that commit. Repos that are level, or merely ahead, are answered out of the local object database with no network at all; repos that really did move get one targeted fetch each.

Two escape hatches: fetch=False leaves unknown counts as ? rather than going to the network, and --offline skips the API entirely for exactly the old pure-git behaviour.

source

local_state

def local_state(
    path
):

Branch, dirty flag and short HEAD of the checkout at path

Exported source 
def git(path, *args):
    "Run git in `path`, returning stripped stdout ('' when it fails)"
    r = subprocess.run(["git", "-C", str(path), *args], capture_output=True, text=True, check=False)
    return r.stdout.strip()

def local_state(path):
    "Branch, dirty flag and short HEAD of the checkout at `path`"
    return {"branch": git(path, "rev-parse", "--abbrev-ref", "HEAD") or None,
            "dirty":  bool(git(path, "status", "--porcelain")),
            "head":   git(path, "rev-parse", "--short=7", "HEAD") or None}

def _have(path, sha):
    "True when `sha` is already a commit here, so no fetch is needed to compare against it"
    return subprocess.run(["git", "-C", str(path), "cat-file", "-e", f"{sha}^{{commit}}"],
                          capture_output=True, check=False).returncode == 0

source

git

def git(
    path, *args
):

Run git in path, returning stripped stdout (’’ when it fails)

source

add_local

def add_local(
    rows, root, fetch:bool=True
):

Merge branch, dirty flag and behind/ahead counts into each row, in place

Exported source 
def ahead_behind(path, remote_sha, branch, fetch=True):
    "(behind, ahead) for `path`, fetching only when the remote commit is not already local"
    if not branch or branch == "HEAD": return (None, None)
    # remote_sha is main's, so it can only stand in for a checkout that is on main
    if branch == "main" and remote_sha and _have(path, remote_sha):
        ref = remote_sha
    elif fetch:
        git(path, "fetch", "-q", "origin")
        ref = f"origin/{branch}"
    else:
        return (None, None)
    counts = git(path, "rev-list", "--left-right", "--count", f"{ref}...HEAD").split()
    return (int(counts[0]), int(counts[1])) if len(counts) == 2 else (None, None)

def add_local(rows, root, fetch=True):
    "Merge branch, dirty flag and behind/ahead counts into each row, in place"
    for r in rows:
        p = Path(root)/r["path"]
        st = local_state(p)
        behind, ahead = ahead_behind(p, r.get("remote"), st["branch"], fetch=fetch)
        r.update(st, behind=behind, ahead=ahead)
    return rows

source

ahead_behind

def ahead_behind(
    path, remote_sha, branch, fetch:bool=True
):

(behind, ahead) for path, fetching only when the remote commit is not already local

def _commit(repo, text):
    (Path(repo)/"f").write_text(text)
    git(repo, "add", ".")
    git(repo, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qm", text)
    return git(repo, "rev-parse", "--short=7", "HEAD")

# No clone and no file:// remote here. `protocol.file.allow=never` is set on some machines
# (knowledge-lab is one), and there `git clone <path>` dies with "transport 'file' not allowed".
# What matters anyway is the no-network path: the whole point of ahead_behind is to answer out of
# the local object database whenever it can, and to say ? rather than guess when it cannot.
with tempfile.TemporaryDirectory() as d:
    repo = Path(d)/"r"
    subprocess.run(["git", "init", "-q", "-b", "main", str(repo)], check=True)
    first  = _commit(repo, "one")
    second = _commit(repo, "two")        # stands in for what the remote's main points at
    git(repo, "tag", "remote", second)   # keep it reachable once main moves back

    st = local_state(repo)
    test_eq(st["branch"], "main")
    test_eq(st["dirty"], False)
    test_eq(st["head"], second)
    test_eq(ahead_behind(repo, second, "main", fetch=False), (0, 0))    # level, no network

    git(repo, "checkout", "-q", "-B", "main", first)                    # HEAD now behind the tag
    test_eq(_have(repo, second), True)
    test_eq(ahead_behind(repo, second, "main", fetch=False), (1, 0))    # answered from local objects

    _commit(repo, "three")                                              # diverged both ways
    test_eq(ahead_behind(repo, second, "main", fetch=False), (1, 1))

    unknown = "0" * 40                   # a sha this checkout has never seen, and no reachable remote
    test_eq(_have(repo, unknown), False)
    test_eq(ahead_behind(repo, unknown, "main", fetch=False), (None, None))
    test_eq(ahead_behind(repo, unknown, "main", fetch=True), (None, None))   # ? beats a wrong 0

    (repo/"g").write_text("x")
    test_eq(local_state(repo)["dirty"], True)
    test_eq(ahead_behind(repo, second, "HEAD", fetch=True), (None, None))    # detached: no counts

The combined table

report and ci are the same rows rendered two ways. ci answers “did what I pushed build?”; report answers “what needs syncing?”, and now flags a failed build while it is there.

source

format_report

def format_report(
    rows, color:bool=True
):

Local sync state plus remote health, one line per submodule

Exported source 
def format_report(rows, color=True):
    "Local sync state plus remote health, one line per submodule"
    def c(s, col): return f"{col}{s}{OFF}" if color and col else s
    head = (f"{'SUBMODULE':<28} {'BRANCH':<9} {'DIRTY':<5} {'BEHIND':<6} "
            f"{'AHEAD':<5} {'CI':<9} {'PAGES':<9} {'BUILT':<7}")
    out = [c(head, BOLD)]
    for r in rows:
        br = r.get("branch") or "?"
        detached = br == "HEAD"
        if detached: br = "DETACHED"
        behind = "?" if r.get("behind") is None else str(r["behind"])
        ahead  = "?" if r.get("ahead")  is None else str(r["ahead"])
        # yellow for a detached HEAD, cyan for anything wanting a sync, push or upload
        local_c = (YELLOW if detached else
                   CYAN if (r.get("dirty") or r.get("behind") or r.get("ahead")) else "")
        ci, ci_c = _state(r["ci"]["state"] if r.get("ci") else None)
        dp, dp_c = _state(r["deploy"]["state"] if r.get("deploy") else None)
        bl, bl_c = _state(r.get("built"))
        out.append(" ".join([c(f"{r['path']:<28}", local_c), c(f"{br:<9}", local_c),
                             c(f"{'YES' if r.get('dirty') else 'no':<5}", local_c),
                             c(f"{behind:<6}", local_c), c(f"{ahead:<5}", local_c),
                             c(f"{ci:<9}", ci_c), c(f"{dp:<9}", dp_c), c(f"{bl:<7}", bl_c)]))
    return "\n".join(out)
full = {**ok, "branch":"main", "dirty":False, "behind":0, "ahead":0}
lines = format_report([full,
                       {**full, "path":"DL/DL_tasks", "dirty":True, "behind":2, "ahead":1},
                       {**full, "path":"ML/ML_methods", "branch":"HEAD", "behind":None, "ahead":None},
                       {**bare, "branch":"main", "dirty":False, "behind":0, "ahead":0}],
                      color=False).splitlines()
test_eq(len(lines), 5)
test_eq(lines[1].split(), ["Web/WEB_doc", "main", "no", "0", "0", "success", "success", "built"])
test_eq(lines[2].split(), ["DL/DL_tasks", "main", "YES", "2", "1", "success", "success", "built"])
test_eq(lines[3].split(), ["ML/ML_methods", "DETACHED", "no", "?", "?", "success", "success", "built"])
test_eq(lines[4].split(), ["Personal/bthek1", "main", "no", "0", "0", "-", "-", "-"])

Auditing .gitmodules

Nothing checks that .gitmodules still describes reality. Two things can drift:

  • a submodule points at a repo that no longer exists under the account, which is real breakage
  • a section name stops matching the path it points at, which is only confusing, but this repo has two of them already ([submodule "DL/Research"] has path = DL/DL_tasks, and [submodule "Personal/Other"] has path = Business/Other)

The third comparison, repos on GitHub that are not submodules here, is informational rather than a finding: the account owns far more repos than this superproject tracks, and almost all of that is deliberate. It is printed as a compact list, not as a to-do.

source

owned_repos

async def owned_repos(
    gh
):

(login, every owned repo, the ones that are neither forks nor archived)

Exported source 
def stale_sections(root=None):
    "[(section name, path)] where a .gitmodules section name no longer matches its path"
    root = Path(root) if root else find_root()
    paths, _ = _gitmodules_kv(root)
    return sorted((n, p) for n, p in paths.items() if n != p)

async def owned_repos(gh):
    "(login, every owned repo, the ones that are neither forks nor archived)"
    login = (await gh.rest.users.async_get_authenticated()).parsed_data.login
    every, active = set(), set()
    async for r in gh.rest.paginate(gh.rest.repos.async_list_for_authenticated_user,
                                    per_page=100, affiliation="owner"):
        every.add(r.name)
        if not (r.fork or r.archived): active.add(r.name)
    return login, every, active

source

stale_sections

def stale_sections(
    root:NoneType=None
):

[(section name, path)] where a .gitmodules section name no longer matches its path

source

gather_audit

async def gather_audit(
    subs, root, token:NoneType=None
):

The audit, one client, one pass over the account’s repos

Exported source 
def self_repo(root):
    "Repo name this superproject itself lives in, so the audit does not list it"
    m = re.search(r"[:/]([^/:]+)/([^/]+?)(?:\.git)?$", git(root, "config", "--get", "remote.origin.url"))
    return m[2] if m else None

def audit_result(subs, login, every, active, stale, exclude=()):
    "Compare .gitmodules against the repos the account owns"
    mine = {s.repo for s in subs if s.owner == login} | {e for e in exclude if e}
    return {"login": login,
            "gone":      sorted(r for r in mine if r not in every),        # real breakage
            "untracked": sorted(r for r in active if r not in mine),       # informational
            "foreign":   sorted(f"{s.owner}/{s.repo}" for s in subs if s.owner != login),
            "stale":     list(stale)}

async def gather_audit(subs, root, token=None):
    "The audit, one client, one pass over the account's repos"
    async with client(token) as gh:
        login, every, active = await owned_repos(gh)
    return audit_result(subs, login, every, active, stale_sections(root),
                        exclude=[self_repo(root)])

source

audit_result

def audit_result(
    subs, login, every, active, stale, exclude:tuple=()
):

Compare .gitmodules against the repos the account owns

source

self_repo

def self_repo(
    root
):

Repo name this superproject itself lives in, so the audit does not list it

source

format_audit

def format_audit(
    res, color:bool=True
):

The audit as a string. Only gone and stale are findings

Exported source 
def _wrap(items, width=92):
    "Comma-joined `items`, wrapped to `width` columns"
    lines, cur = [], ""
    for it in items:
        nxt = it if not cur else f"{cur}, {it}"
        if len(nxt) > width and cur: lines.append(cur + ","); cur = it
        else: cur = nxt
    if cur: lines.append(cur)
    return lines

def format_audit(res, color=True):
    "The audit as a string. Only `gone` and `stale` are findings"
    def c(s, col): return f"{col}{s}{OFF}" if color and col else s
    out = [c(f"Audit of .gitmodules against github.com/{res['login']}", BOLD)]
    if res["gone"]:
        out.append(c(f"\nIn .gitmodules but not on GitHub ({len(res['gone'])}):", RED))
        out += [f"  {r}" for r in res["gone"]]
    if res["stale"]:
        out.append(c(f"\nSection name no longer matches its path ({len(res['stale'])}):", YELLOW))
        out += [f'  [submodule "{n}"] -> path = {p}' for n, p in res["stale"]]
    if not res["gone"] and not res["stale"]:
        out.append("\nEvery submodule resolves to a repo that exists, and every section name matches.")
    if res["foreign"]:
        out.append(c(f"\nSubmodules owned by someone else ({len(res['foreign'])}):", DIM))
        out += [f"  {r}" for r in res["foreign"]]
    out.append(c(f"\nOwned repos that are not submodules here ({len(res['untracked'])}), "
                 f"usually deliberate:", DIM))
    out += [f"  {line}" for line in _wrap(res["untracked"])]
    return "\n".join(out)
subs_fx = [Submodule("DL/DL_tasks", "bthek1", "DL_tasks"),
           Submodule("Personal/bthek1", "bthek1", "bthek1"),
           Submodule("Vendor/thing", "someone", "thing")]

res = audit_result(subs_fx, "bthek1", {"DL_tasks", "bthek1", "Old"}, {"DL_tasks", "bthek1", "Old"},
                   [("DL/Research", "DL/DL_tasks")])
test_eq(res["gone"], [])
test_eq(res["untracked"], ["Old"])
test_eq(res["foreign"], ["someone/thing"])          # not compared against the account's repos
test_eq(res["stale"], [("DL/Research", "DL/DL_tasks")])

gone = audit_result(subs_fx, "bthek1", {"bthek1"}, {"bthek1"}, [])
test_eq(gone["gone"], ["DL_tasks"])
assert "not on GitHub" in format_audit(gone, color=False)

# archived and forked repos are owned but not active, so they are never proposed as untracked
test_eq(audit_result([], "bthek1", {"Live", "Archived"}, {"Live"}, [])["untracked"], ["Live"])
assert "matches" in format_audit(audit_result([], "bthek1", set(), set(), []), color=False)

with tempfile.TemporaryDirectory() as d:
    (Path(d)/".gitmodules").write_text(FIXTURE)
    test_eq(stale_sections(d), [("DL/Research", "DL/DL_tasks")])   # Personal/bthek1 matches its path

test_eq(_wrap(["aaa", "bbb", "ccc"], width=8), ["aaa, bbb,", "ccc"])

# the superproject's own repo is owned but is not a submodule of itself: excluded, not "untracked"
excl = audit_result([], "bthek1", {"Knowledge", "Other"}, {"Knowledge", "Other"}, [], exclude=["Knowledge"])
test_eq(excl["untracked"], ["Other"])
test_eq(excl["gone"], [])

Entry point

ghstatus is the console script the justfile calls (just ci, and just report --report). Exit codes carry meaning, because report acts on them:

Code Means What just report does
0 asked, everything healthy nothing
1 asked, something failed nothing; the failing URLs are printed
2 bad input (no .gitmodules under --root) gives up; a fallback cannot help
3 cannot ask (no token, no githubkit, rate limited, offline) falls back to the pure-git report

With --audit, 1 means a submodule points at a repo that no longer exists.

source

ghstatus

def ghstatus(
    report:bool=False, # Add the local columns: branch, dirty, behind, ahead
    audit:bool=False, # Audit .gitmodules against the repos the account owns
    offline:bool=False, # Skip the GitHub API entirely (implies --report)
    no_fetch:bool=False, # Never fetch; unknown behind/ahead counts stay as ?
    json:bool=False, # Emit JSON instead of the table
    no_color:bool=False, # Disable ANSI colour
    root:str=None, # Superproject root (default: nearest ancestor with .gitmodules)
):

Remote CI, Pages and main-branch status for every submodule

Exported source 
@call_parse
def ghstatus(
    report:bool=False,    # Add the local columns: branch, dirty, behind, ahead
    audit:bool=False,     # Audit .gitmodules against the repos the account owns
    offline:bool=False,   # Skip the GitHub API entirely (implies --report)
    no_fetch:bool=False,  # Never fetch; unknown behind/ahead counts stay as ?
    json:bool=False,      # Emit JSON instead of the table
    no_color:bool=False,  # Disable ANSI colour
    root:str=None,        # Superproject root (default: nearest ancestor with .gitmodules)
):
    "Remote CI, Pages and main-branch status for every submodule"
    import sys
    from json import dumps
    try:
        base = Path(root) if root else find_root()
        subs = parse_gitmodules(base)
        if audit:
            res = asyncio.run(gather_audit(subs, base))
        elif offline:
            rows = [{"path": s.path, "owner": s.owner, "repo": s.repo, "ci": None, "deploy": None,
                     "built": None, "remote": None, "local": local_sha(base/s.path)} for s in subs]
        else:
            rows = asyncio.run(gather_status(subs, root=base))
        if report or offline: add_local(rows, base, fetch=not no_fetch)
    except FileNotFoundError as e:      # 2: bad input, and no fallback would help
        print(f"{RED}{e}{OFF}", file=sys.stderr)
        sys.exit(2)
    except (ImportError, RuntimeError) as e:   # 3: cannot ask, so a git-only view is still useful
        print(f"{RED}{e}{OFF}", file=sys.stderr)
        sys.exit(3)
    except Exception as e:              # 3: auth, rate limit, transport. Never report health here
        print(f"{RED}{api_error(e)}{OFF}", file=sys.stderr)
        sys.exit(3)
    if audit:
        print(dumps(res, indent=2) if json else format_audit(res, color=not no_color))
        sys.exit(1 if res["gone"] else 0)
    if json:
        print(dumps({r["path"]: r for r in rows}, indent=2))
        sys.exit(0)
    print(format_report(rows, color=not no_color) if (report or offline)
          else format_table(rows, color=not no_color))
    for path, kind, url in failures(rows):
        print(f"{RED}{path}{OFF} {kind} failed: {url}")
    hint = ("BEHIND>0 -> just sync   AHEAD>0 -> just push   DIRTY -> just upload"
            if (report or offline) else "LOCAL != REMOTE -> just sync / just push")
    print(f"\n{DIM}FAILURE -> open the url above   {hint}{OFF}")
    sys.exit(1 if has_failure(rows) else 0)

python -m nbdevAuto.github 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.

Live, against the real superproject. Marked eval: false because CI has no gh login.

rows = asyncio.run(gather_status(parse_gitmodules("/home/bthek1/Knowledge"), root="/home/bthek1/Knowledge"))
print(format_table(rows))
Back to top