This commit is contained in:
Waylon Walker 2025-11-25 11:57:40 -06:00
parent 7944f15f61
commit 88eb5122ab

173
workspaces.py Normal file → Executable file
View file

@ -12,6 +12,7 @@
import os import os
import re import re
import shutil
import subprocess import subprocess
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@ -22,7 +23,7 @@ from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from rich.console import Console from rich.console import Console
from rich.table import Table from rich.table import Table
from rich.prompt import Prompt from rich.prompt import Prompt, Confirm
from iterfzf import iterfzf from iterfzf import iterfzf
app = typer.Typer( app = typer.Typer(
@ -60,7 +61,6 @@ class Settings(BaseSettings):
env_prefix="", env_prefix="",
env_file=None, env_file=None,
env_nested_delimiter="__", env_nested_delimiter="__",
# map env vars explicitly
extra="ignore", extra="ignore",
) )
@ -72,12 +72,9 @@ class Settings(BaseSettings):
2. WORKSPACES_NAME env 2. WORKSPACES_NAME env
3. default "git" 3. default "git"
""" """
# First, load from env (WORKSPACES_NAME)
s = cls() s = cls()
if override_workspaces_name is not None: if override_workspaces_name is not None:
s.workspaces_name = override_workspaces_name s.workspaces_name = override_workspaces_name
# We still want WORKSPACES_NAME to be honored even though
# we don't use "fields" config anymore:
env_val = os.getenv("WORKSPACES_NAME") env_val = os.getenv("WORKSPACES_NAME")
if env_val and override_workspaces_name is None: if env_val and override_workspaces_name is None:
s.workspaces_name = env_val s.workspaces_name = env_val
@ -287,7 +284,6 @@ def main(
# Default behavior when no subcommand is provided: # Default behavior when no subcommand is provided:
if ctx.invoked_subcommand is None: if ctx.invoked_subcommand is None:
# Call the list_workspaces command programmatically
list_workspaces(ctx) list_workspaces(ctx)
raise typer.Exit(0) raise typer.Exit(0)
@ -346,6 +342,7 @@ def list_workspaces(
@app.command("create") @app.command("create")
@app.command("new", hidden=True)
def create_workspace( def create_workspace(
ctx: typer.Context, ctx: typer.Context,
name: Optional[str] = typer.Option( name: Optional[str] = typer.Option(
@ -569,6 +566,168 @@ def add_repo(
) )
# ---------------- rm-workspace ----------------
def find_repo_for_worktree(
worktree_path: Path, repos_dir: Path
) -> Optional[Path]:
"""
Try to find the parent repo for a worktree, assuming it lives in repos_dir
with the same directory name as the worktree.
"""
candidate = repos_dir / worktree_path.name
if candidate.exists() and ensure_git_repo(candidate):
return candidate
return None
def has_unpushed_commits(repo_path: Path, branch: str) -> bool:
"""
Detect if branch has commits not on its upstream.
Uses 'git rev-list @{u}..HEAD'; if non-empty, there are unpushed commits.
"""
# Check if upstream exists
code, _, _ = run_cmd(
["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"],
cwd=repo_path,
)
if code != 0:
# No upstream configured; treat as unpushed work.
return True
code, out, _ = run_cmd(
["git", "rev-list", "@{u}..HEAD", "--count"],
cwd=repo_path,
)
if code != 0:
return True
try:
count = int(out.strip() or "0")
except ValueError:
return True
return count > 0
@app.command("rm")
def remove_workspace(
ctx: typer.Context,
workspace: Optional[str] = typer.Option(
None,
"--workspace",
"-w",
help=(
"Workspace to remove. "
"If omitted, uses the workspace containing the current directory."
),
),
force: bool = typer.Option(
False,
"--force",
"-f",
help="Force removal even if there is dirty or unpushed work.",
),
):
"""
Remove a workspace:
- For each repo worktree in the workspace:
* Check for dirty work or unpushed commits.
* If any found and not --force, abort.
* Otherwise, run 'git worktree remove'.
- Finally, delete the workspace directory.
"""
_settings, repos_dir, workspaces_dir = get_ctx_paths(ctx)
ws_dir = find_workspace_dir(workspaces_dir, workspace)
if not ws_dir.exists():
console.print(f"[red]Workspace '{ws_dir.name}' does not exist at {ws_dir}[/red]")
raise typer.Exit(1)
# Collect worktrees (subdirs that look like git repos, excluding readme.md)
worktrees: List[Path] = []
for child in sorted(p for p in ws_dir.iterdir() if p.is_dir()):
if child.name.lower() == "readme.md":
continue
if not ensure_git_repo(child):
continue
worktrees.append(child)
if not worktrees:
# No worktrees, just remove workspace dir
if not Confirm.ask(
f"[red]No worktrees found.[/red] Remove empty workspace [bold]{ws_dir.name}[/bold]?",
default=False,
):
raise typer.Exit(0)
shutil.rmtree(ws_dir)
console.print(f"[green]Removed workspace[/green] {ws_dir}")
raise typer.Exit(0)
# Check for dirty / unpushed changes
problems: List[str] = []
for wt in worktrees:
status = get_git_status(wt)
branch = get_current_branch(wt) or "?"
if status.dirty:
problems.append(f"{wt.name}: dirty working tree on '{branch}'")
repo = find_repo_for_worktree(wt, repos_dir)
if repo is not None and branch != "?":
if has_unpushed_commits(wt, branch):
problems.append(f"{wt.name}: unpushed commits on '{branch}'")
if problems and not force:
console.print(
"[red]Refusing to remove workspace; found dirty or unpushed work:[/red]"
)
for p in problems:
console.print(f" - {p}")
console.print(
"\nUse [bold]--force[/bold] to remove the workspace and its worktrees anyway."
)
raise typer.Exit(1)
if not Confirm.ask(
f"Remove workspace [bold]{ws_dir.name}[/bold] and clean up its worktrees?",
default=False,
):
raise typer.Exit(0)
# Remove worktrees via git
for wt in worktrees:
repo = find_repo_for_worktree(wt, repos_dir)
display_name = wt.name
if repo is None:
console.print(
f"[yellow]Could not find parent repo for worktree {display_name}; "
"deleting directory directly.[/yellow]"
)
shutil.rmtree(wt)
continue
console.print(f"Removing worktree for [bold]{display_name}[/bold]...")
code, _out, err = run_cmd(
["git", "worktree", "remove", "--force" if force else "--detach", str(wt)],
cwd=repo,
)
if code != 0:
console.print(
f"[red]Failed to remove worktree {display_name} via git:[/red]\n{err}"
)
# As a last resort, if force is given, nuke the dir.
if force:
console.print(
f"[yellow]Forcing directory removal of {wt} despite git error.[/yellow]"
)
shutil.rmtree(wt)
elif wt.exists():
# git worktree remove should delete the directory; if not, clean up.
shutil.rmtree(wt)
# Finally, remove the workspace directory itself
shutil.rmtree(ws_dir)
console.print(f"[green]Removed workspace[/green] {ws_dir}")
if __name__ == "__main__": if __name__ == "__main__":
app() app()