render_precommit¶

def chain(references: list[str], relative_to: Path, layout: Layout, seen: list[Path] | None = None) -> list[Fragment]:
    """Expand fragment references into the flat, deduplicated list to merge.

    Args:
        references: The fragment references, in merge order.
        relative_to: The directory references are resolved against first.
        layout: Where the repository root and fragment directory are.
        seen: Paths already in the chain, threaded through the recursion to
            deduplicate a shared base and to catch a cycle.

    Returns:
        The fragments in merge order, each appearing once, dependencies first.

    Raises:
        FragmentError: If a reference cannot be resolved or ``extends`` forms a cycle.
    """
    seen = [] if seen is None else seen
    result: list[Fragment] = []
    for reference in references:
        path = resolve(reference, relative_to, layout)
        if path in seen:
            continue
        seen.append(path)
        fragment = parse_fragment(path)
        result.extend(chain(fragment.extends, path.parent, layout, seen))
        result.append(fragment)
    return result

def deployable(layout: Layout) -> list[Path]:
    """List the fragments that declare an output, i.e. those rendered by default.

    Args:
        layout: Where the fragment directory is.

    Returns:
        The fragment paths, sorted. A fragment with no ``output`` is a mixin and is
        skipped -- it is only ever pulled in via ``extends``.

    Raises:
        FragmentError: If the fragment directory is missing.
    """
    if not layout.fragment_dir.is_dir():
        msg = f"no fragment directory at {layout.display(layout.fragment_dir)}"
        raise FragmentError(msg)
    return sorted(path for path in layout.fragment_dir.glob("*.y*ml") if parse_fragment(path).output)

def header(fragments: list[Fragment], layout: Layout) -> list[str]:
    """Build the rendered file's header comment.

    Args:
        fragments: The chain the file was rendered from, in merge order.
        layout: Where the repository root is, so fragments are named relative to it.

    Returns:
        The header lines, naming the fragments so whoever opens the deployed config
        is sent to the right place to edit it.
    """
    names = ", ".join(layout.display(fragment.path) for fragment in fragments)
    return [
        "# GENERATED FILE - do not edit. Rendered by rhiza-hooks (render-precommit)",
        f"# from {names}.",
        "# Edit a fragment and re-render.",
        "#",
        "# pre-commit hard-codes this path, so the language layers that deploy it are",
        "# alternatives rather than files that could coexist.",
    ]

def main(argv: list[str] | None = None) -> int:
    """Render the requested configs, or check the deployed files for drift.

    Args:
        argv: Command-line arguments, defaulting to ``sys.argv[1:]``.

    Returns:
        A process exit status: 0 on success, 1 if a check found drift or a fragment
        could not be merged.
    """
    args = _build_parser().parse_args(argv)
    layout = Layout.discover(fragment_dir=args.fragment_dir)

    # A repository that renders nothing is the normal case, and a shared build step or CI
    # job may call this unconditionally across a fleet of them. So an absent *default*
    # fragment directory is "nothing to do", not an error. An explicit --fragment-dir is
    # an assertion that fragments live there, and a missing one still fails below.
    if not args.fragments and args.fragment_dir == DEFAULT_FRAGMENT_DIR and not layout.fragment_dir.is_dir():
        print(f"no {DEFAULT_FRAGMENT_DIR}/ directory - nothing to render")
        return 0

    try:
        targets = [args.fragments] if args.fragments else [[str(path)] for path in deployable(layout)]
        if not targets:
            print(f"error: no fragment in {layout.display(layout.fragment_dir)}/ declares an output:", file=sys.stderr)
            return 1
        rendered = plan(targets, args.out, set(args.exclude_hook), set(args.exclude_repo), layout)
    except (FragmentError, OSError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    changed = sum(_apply(path, text, args.write, layout) for path, text in rendered)
    if not args.write and changed:
        print(f"\n{changed} config(s) out of date. Re-render with --write.", file=sys.stderr)
        return 1
    return 0

def merge(
    fragments: list[Fragment], extra_hooks: set[str], extra_repos: set[str]
) -> tuple[dict[str, Block], dict[str, Repo]]:
    """Merge a chain of fragments into one config's blocks.

    Args:
        fragments: The fragments, in merge order.
        extra_hooks: Hook ids to remove on top of what the fragments declare.
        extra_repos: Repository URLs to remove on top of what the fragments declare.

    Returns:
        A ``(preamble, repos)`` pair holding the merged blocks in output order, with
        every removal applied.

    Raises:
        FragmentError: If two fragments pin one repository to different revs.
    """
    preamble: dict[str, Block] = {}
    repos: dict[str, Repo] = {}
    drop_hooks = set(extra_hooks)
    drop_repos = set(extra_repos)
    for fragment in fragments:
        preamble.update(fragment.preamble)
        drop_hooks |= fragment.remove_hooks
        drop_repos |= fragment.remove_repos
        for url, repo in fragment.repos.items():
            if url in repos:
                _absorb(repos[url], repo, fragment.path)
            else:
                repos[url] = repo.copy()
    for url in drop_repos:
        repos.pop(url, None)
    for repo in repos.values():
        repo.hooks = [hook for hook in repo.hooks if hook.hook_id not in drop_hooks]
    # A repository whose every hook was removed would emit an entry with an empty
    # hooks: list, which pre-commit rejects.
    return preamble, {url: repo for url, repo in repos.items() if repo.hooks}

def parse_fragment(path: Path) -> Fragment:
    """Parse a fragment file into its meta keys, preamble blocks and repositories.

    Args:
        path: The fragment to read.

    Returns:
        The parsed fragment.

    Raises:
        FragmentError: If the file has no top-level keys, or a meta key has the wrong
            shape.
    """
    lines = path.read_text(encoding="utf-8").splitlines()
    starts = [index for index, line in enumerate(lines) if line and not line[0].isspace() and not _is_comment(line)]
    if not starts:
        msg = f"{path}: no top-level keys found"
        raise FragmentError(msg)
    fragment = Fragment(path=path)
    for index, start in enumerate(starts):
        end = starts[index + 1] if index + 1 < len(starts) else len(lines)
        block = Block(
            comments=_key_comments(lines, start, starts[index - 1] if index else 0),
            body=_rstrip(lines[start:end]),
        )
        key = lines[start].split(":", 1)[0].strip()
        if key in META_KEYS:
            _apply_meta(fragment, key, _meta(block, path), path)
        elif key == "repos":
            fragment.repos = _parse_repos(block.body[1:], path)
        else:
            fragment.preamble[key] = block
    return fragment

def plan(
    targets: list[list[str]], out: str | None, hooks: set[str], repos: set[str], layout: Layout
) -> list[tuple[Path, str]]:
    """Render every target chain, resolving where each result is deployed.

    Args:
        targets: Fragment reference lists, one per config to render.
        out: An explicit destination overriding the chain's ``output:``.
        hooks: Hook ids to remove on top of what the fragments declare.
        repos: Repository URLs to remove on top of what the fragments declare.
        layout: Where the repository root and fragment directory are.

    Returns:
        The ``(path, text)`` pairs to write, in target order.

    Raises:
        FragmentError: If a chain cannot be resolved or merged, if it declares no
            destination and none was given, or if the destination is one of its own
            inputs.
    """
    rendered: list[tuple[Path, str]] = []
    for references in targets:
        fragments = chain(references, Path.cwd(), layout)
        destination = out or fragments[-1].output
        if not destination:
            msg = f"{layout.display(fragments[-1].path)} declares no output: -- pass --out"
            raise FragmentError(msg)
        target = (layout.repo_root / destination).resolve()
        _reject_self_reference(target, fragments, layout)
        rendered.append((target, render(fragments, layout, hooks, repos)))
    return rendered

def render(
    fragments: list[Fragment],
    layout: Layout,
    # AbstractSet, not set: these are read-only here (merge copies them), and a
    # mutable default would be a shared-state bug waiting to happen.
    extra_hooks: AbstractSet[str] = frozenset(),
    extra_repos: AbstractSet[str] = frozenset(),
) -> str:
    """Render a chain of fragments into a complete ``.pre-commit-config.yaml``.

    Args:
        fragments: The fragments, in merge order.
        layout: Where the repository root is.
        extra_hooks: Hook ids to remove on top of what the fragments declare.
        extra_repos: Repository URLs to remove on top of what the fragments declare.

    Returns:
        The file's full text, newline-terminated.

    Raises:
        FragmentError: If the merge conflicts, or the result is not a valid config.
    """
    preamble, repos = merge(fragments, set(extra_hooks), set(extra_repos))
    lines = header(fragments, layout)
    for block in preamble.values():
        lines.extend(["", *block.render()])
    lines.extend(["", "repos:"])
    for index, repo in enumerate(repos.values()):
        if index:
            lines.append("")
        lines.extend(repo.render())
    text = "\n".join(lines) + "\n"
    _validate(text, fragments[-1].path)
    return text

def resolve(reference: str, relative_to: Path, layout: Layout) -> Path:
    """Resolve a fragment reference to a path.

    Args:
        reference: A bare name (``python`` or ``python.yaml``), resolved inside the
            fragment directory, or a path, resolved against the referring fragment's
            directory and then the repository root.
        relative_to: The directory of the fragment doing the referring.
        layout: Where the repository root and fragment directory are.

    Returns:
        The resolved path.

    Raises:
        FragmentError: If no candidate exists.
    """
    name = reference if reference.endswith((".yaml", ".yml")) else f"{reference}.yaml"
    candidates = (
        [Path(name)]
        if Path(name).is_absolute()
        else [relative_to / name, layout.repo_root / name, layout.fragment_dir / name]
    )
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    # dict.fromkeys, not set(): the candidates collapse to one entry when the referring
    # fragment sits at the repository root, and the order they were tried in is the
    # useful part of the message.
    tried = ", ".join(dict.fromkeys(layout.display(candidate) for candidate in candidates))
    msg = f"no fragment {reference!r} (tried: {tried})"
    raise FragmentError(msg)