Add copier-python CLI with update subcommand

.github/workflows/ci.yaml

@@ -26,6 +26,8 @@ jobs:
     steps:
       - name: 💾 Check out repository
         uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
 
       - name: 🪪 Configure git identity for tests
         run: |

copier_python/__main__.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import os
+import shutil
+from dataclasses import dataclass
+from enum import Enum
+from typing import TYPE_CHECKING, Annotated
+
+from rich import print  # noqa: A004
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+from rich.text import Text
+from typer import Argument, Context, Exit, Option, Typer
+
+from .repo import RepoTarget
+from .update import UpdateAction
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+
+console = Console()
+
+
+class Args:
+    Repos = Annotated[
+        list[str],
+        Argument(
+            help=(
+                "Repositories to update. Accepts gh:user/repo"
+                " or github.com/user/repo."
+            )
+        ),
+    ]
+    DryRun = Annotated[
+        bool, Option("--dry-run", "-n", help="Skip push and PR creation.")
+    ]
+
+
+cli = Typer(
+    help="copier-python utilities",
+    add_completion=False,
+    no_args_is_help=True,
+    pretty_exceptions_enable=False,
+)
+
+
+@cli.callback()
+def setup(ctx: Context) -> None:
+    pass
+
+
+class UpdateStatus(Enum):
+    UPDATED = "bold green"
+    CURRENT = "bold blue"
+    FAILED = "bold red"
+
+    @property
+    def formatted_name(self) -> Text:
+        return Text(f"{self.name:<7}", style=self.value)
+
+
+@dataclass
+class UpdateResult:
+    repo: RepoTarget
+    status: UpdateStatus
+    exception: Exception | None = None
+    pr_url: str | None = None
+
+
+@cli.command()
+def update(
+    repos: Args.Repos,
+    *,
+    dry_run: Args.DryRun = False,
+    branch: Annotated[str, Option(help="Branch name to create.")] = "updates",
+) -> None:
+    """Apply copier-python template updates to one or more downstream repos."""
+    results = []
+
+    repo_targets = {(target := RepoTarget(repo)).url: target for repo in repos}
+    for target in repo_targets.values():
+        try:
+            pr_url = UpdateAction(target, branch=branch, dry_run=dry_run)()
+            if pr_url:
+                results.append(
+                    UpdateResult(target, UpdateStatus.UPDATED, pr_url=pr_url)
+                )
+            else:
+                results.append(UpdateResult(target, UpdateStatus.CURRENT))
+        except Exception as exc:  # noqa: BLE001, PERF203
+            console.print_exception()
+            results.append(
+                UpdateResult(target, status=UpdateStatus.FAILED, exception=exc)
+            )
+
+    _print_summary(results, dry_run=dry_run)
+    if any(r.status == UpdateStatus.FAILED for r in results):
+        raise Exit(1)
+
+
+def _print_summary(results: Sequence[UpdateResult], *, dry_run: bool) -> None:
+    if not results:
+        return
+    grid = Table.grid(padding=(0, 1), expand=True)
+    grid.add_column()
+    grid.add_column()
+    grid.add_column()
+    for result in sorted(results, key=lambda r: r.repo.github_repo):
+        grid.add_row(
+            result.status.formatted_name,
+            Text.from_markup(
+                f"[link={result.repo.url}]{result.repo.github_repo}[/link]",
+                style="bold",
+            ),
+            Text(str(result.pr_url or result.exception)),
+        )
+    title = Text("Update Results", style="bold")
+    if dry_run:
+        title += Text(" (dry run)", style="green")
+    print(
+        Panel(grid, title=title, title_align="left", expand=False, padding=1)
+    )
+
+
+def setup_env() -> None:
+    os.environ.pop("VIRTUAL_ENV", default=None)
+    os.environ["TERMINAL_WIDTH"] = str(
+        min(shutil.get_terminal_size().columns, 100)
+    )
+
+
+def main() -> None:
+    setup_env()
+    cli()
+
+
+if __name__ == "__main__":
+    main()

copier_python/repo.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import tempfile
+from contextlib import contextmanager, suppress
+from dataclasses import InitVar, dataclass, field
+from functools import cached_property
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar
+
+import yaml
+from rich import print  # noqa: A004
+from rich.panel import Panel
+from rich.text import Text
+from typing_extensions import Self
+
+if TYPE_CHECKING:
+    from collections.abc import Generator, Sequence
+
+
+@dataclass(unsafe_hash=True)
+class RepoTarget:
+    arg: InitVar[str | Path]
+    github_repo: str = field(init=False)
+
+    _GITHUB_REGEXES: ClassVar[Sequence[str]] = [
+        r"(\.git)?/?$",
+        r"^gh\:",
+        r"^(https?://)?github.com/",
+        r"^((git\+)?ssh://)?([a-zA-Z0-9]+)@github.com:",
+    ]
+
+    def __post_init__(self, arg: str | Path) -> None:
+        arg = str(arg).strip()
+        if ((path := Path(arg)).is_dir()) and (root := self.repo_root(path)):
+            arg = root
+        for regex in self._GITHUB_REGEXES:
+            arg = re.sub(regex, "", arg)
+        if not re.fullmatch(r"[\w-]+\/[\w-]+", arg):
+            raise ValueError(arg)
+        self.github_repo = arg
+
+    def repo_root(self, path: Path) -> str | None:
+        with suppress(subprocess.CalledProcessError):
+            return subprocess.check_output(
+                ["git", "config", "--local", "remote.origin.url"],
+                text=True,
+                cwd=path,
+            ).strip()
+        return None
+
+    @cached_property
+    def name(self) -> str:
+        return self.github_repo.split("/")[-1]
+
+    @property
+    def url(self) -> str:
+        return f"https://github.com/{self.github_repo}"
+
+    @cached_property
+    def push_url(self) -> str:
+        return f"git@github.com:{self.github_repo}.git"
+
+
+@dataclass
+class RepoWorktree:
+    path: Path
+    repo: RepoTarget
+    branch: str
+
+    @classmethod
+    @contextmanager
+    def clone(
+        cls, repo: RepoTarget, branch: str
+    ) -> Generator[Self, None, None]:
+        with tempfile.TemporaryDirectory() as td:
+            repo_dir = Path(td) / "worktree"
+            cls.run_in(["git", "clone", repo.url, str(repo_dir)], repo=repo)
+            for cmd in (
+                [
+                    *["git", "remote", "set-url", "--push", "origin"],
+                    repo.push_url,
+                ],
+                ["poe", "setup"],
+                ["git", "checkout", "-b", branch],
+            ):
+                cls.run_in(cmd, repo=repo, cwd=repo_dir)
+            yield cls(path=repo_dir, repo=repo, branch=branch)
+
+    @classmethod
+    def run_in(
+        cls, cmd: list[str], *, repo: RepoTarget, **kwargs: Any
+    ) -> subprocess.CompletedProcess[str]:
+        kwargs.setdefault("check", True)
+        kwargs.setdefault("text", True)
+        cmd_text = Text(" ".join(cmd), style="color(153)")
+        txt = Text.assemble(
+            Text(f"{repo.github_repo} => ", style="bold"),
+            cmd_text,
+        )
+        print(Panel(txt, expand=False, border_style="white dim"))
+        return subprocess.run(cmd, **kwargs)  # noqa: S603 PLW1510
+
+    def run(
+        self, cmd: list[str], **kwargs: Any
+    ) -> subprocess.CompletedProcess[str]:
+        kwargs.setdefault("cwd", self.path)
+        return self.run_in(cmd, repo=self.repo, **kwargs)
+
+    def git_status(self) -> list[str]:
+        return self.run(
+            ["git", "status", "--porcelain"], capture_output=True
+        ).stdout.splitlines()
+
+    @staticmethod
+    def has_conflicts(status: list[str]) -> bool:
+        conflict_codes = {"UU", "AA", "DD", "AU", "UA", "DU", "UD"}
+        return any(line[:2] in conflict_codes for line in status)
+
+    @property
+    def template_ref(self) -> str:
+        return yaml.safe_load(  # type: ignore[no-any-return]
+            (self.path / ".copier-answers.yml").read_text()
+        ).get("_commit")
+
+    def shell(self) -> None:
+        self.run([os.environ.get("SHELL", "/bin/bash")], check=False)
+
+    def open_pr(self, title: str, body: str) -> str:
+        result = self.run(
+            [
+                "gh",
+                "pr",
+                "create",
+                "--title",
+                title,
+                "--body",
+                body,
+                "--head",
+                self.branch,
+            ],
+            capture_output=True,
+            check=False,
+        )
+        if result.returncode == 0:
+            return result.stdout.strip()
+        view = self.run(
+            ["gh", "pr", "view", "--json", "url", "--jq", ".url"],
+            capture_output=True,
+            check=False,
+        )
+        if view.returncode == 0:
+            return view.stdout.strip()
+        raise RuntimeError(f"gh pr create failed: {result.stderr.strip()}")

copier_python/update.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from .repo import RepoWorktree
+
+if TYPE_CHECKING:
+    from .repo import RepoTarget
+
+
+@dataclass
+class UpdateAction:
+    repo: RepoTarget
+    branch: str
+    dry_run: bool = False
+
+    def __call__(self) -> str | None:
+        with RepoWorktree.clone(self.repo, branch=self.branch) as worktree:
+            return self._run(worktree)
+
+    def _run(self, repo: RepoWorktree) -> str | None:
+        """Run copier update in the repo worktree."""
+        copier_status = json.loads(
+            repo.run(
+                ["copier", "check-update", "--output-format", "json"],
+                capture_output=True,
+            ).stdout.strip()
+        )
+        start_ref = "v" + copier_status["current_version"]
+        end_ref = "v" + copier_status["latest_version"]
+        if not copier_status.get("update_available", False):
+            return None
+        repo.run(["copier", "update", "-l"])
+
+        status = repo.git_status()
+        if not status:
+            return None
+
+        if repo.has_conflicts(status):
+            print(  # noqa: T201
+                "Conflicts detected."
+                " Resolve them and exit the shell to continue."
+            )
+            repo.shell()
+            status = repo.git_status()
+            if repo.has_conflicts(status):
+                raise RuntimeError("Conflicts remain, aborting")
+
+        try:
+            repo.run(["uv", "run", "poe", "lt"])
+        except subprocess.CalledProcessError:
+            print(  # noqa: T201
+                "Lint/test failed. Fix errors and exit the shell to continue."
+            )
+            repo.shell()
+
+        title = "Apply template updates"
+        body = ""
+        if start_ref and end_ref and start_ref != end_ref:
+            ref_range = f"{start_ref}...{end_ref}"
+            body = os.linesep.join(
+                (
+                    f"Applied updates from template: {ref_range}",
+                    f"{repo.repo.url}/compare/{ref_range}",
+                )
+            )
+        repo.run(["git", "add", "-A"])
+        repo.run(["git", "commit", "-m", f"{title}\n\n{body}".strip()])
+
+        if self.dry_run:
+            return None
+
+        repo.run(
+            ["git", "push", "-u", "origin", repo.branch, "--force-with-lease"]
+        )
+        return repo.open_pr(title, body)