eblume/blumeops
1
1
Fork 0
Code Issues Pull requests Projects Releases 83 Packages Wiki Activity Actions

Use Forgejo API for branch deletion, add scheduled workflow

Browse source 
- Switch remote branch deletion from git push to Forgejo API
- Token resolved from 1Password locally, --token flag for CI
- Add branch-cleanup.yaml workflow: runs ~every 10 days on schedule,
  supports manual dispatch with configurable cutoff
- Shallow checkout sufficient (no fetch-depth: 0 needed)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Erich Blume 2026-02-22 15:54:38 -08:00
commit ddc809f1f9
2 changed files with 282 additions and 185 deletions
Download patch file Download diff file Expand all files Collapse all files

41
.forgejo/workflows/branch-cleanup.yaml Normal file
View file

@ -0,0 +1,41 @@
# Automated Branch Cleanup
#
# Deletes remote branches that have been merged into main and are older
# than a cutoff (default 30 days). Detects both fast-forward and
# squash-merged branches via the Forgejo API.
#
# Runs on a schedule (~every 10 days) and can be triggered manually
# with a custom cutoff for testing.
name: Branch Cleanup
on:
schedule:
# Approximately every 10 days: 1st, 11th, 21st of each month at 06:00 UTC
- cron: '0 6 1,11,21 * *'
workflow_dispatch:
inputs:
cutoff:
description: 'Delete branches older than N days'
required: false
default: '30'
type: string
jobs:
cleanup:
runs-on: k8s
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Run branch cleanup
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
CUTOFF="${{ inputs.cutoff || '30' }}"
echo "Running branch cleanup with cutoff=${CUTOFF} days..."
uv run --script mise-tasks/branch-cleanup \
--remote-only \
--yes \
--token "$GITHUB_TOKEN" \
--cutoff "$CUTOFF"

416
mise-tasks/branch-cleanup
View file

@ -9,6 +9,7 @@
#USAGE flag "--yes" help="Skip confirmation prompt"
#USAGE flag "--local-only" help="Only clean up local branches"
#USAGE flag "--remote-only" help="Only clean up remote branches"
#USAGE flag "--token <token>" help="Forgejo API token (default: read from 1Password)"
#USAGE flag "--cutoff <cutoff>" default="30" help="Only delete branches whose HEAD commit is older than N days (default 30)"
"""Clean up merged branches locally and on the Forgejo remote.
@ -16,9 +17,11 @@ Detects merged branches via two methods:
1. git branch --merged (catches fast-forward merges)
2. Forgejo API (catches squash-merged PRs)
For remote branches, deletes those confirmed merged by either method.
For local branches, deletes those that match a merged remote branch or
are detected as merged by git.
Remote branches are deleted via the Forgejo API. The token is resolved:
1. --token flag (for CI: pass $GITHUB_TOKEN)
2. 1Password: op read (for local use, prompts biometric)
Local branches are deleted via git branch -D.
Warns about stale local branches that couldn't be confirmed as merged.
@ -40,7 +43,9 @@ from rich.table import Table
PROTECTED_BRANCHES = {"main", "master"}
FORGE_API = "https://forge.ops.eblu.me/api/v1"
REPO = "eblume/blumeops"
REPO_OWNER = "eblume"
REPO_NAME = "blumeops"
OP_TOKEN_REF = "op://vg6xf6vvfmoh5hqjjhlhbeoaie/w3663ffnvkewbftncqxtcpeavy/api-token"
def run_git(*args: str) -> str:
@ -54,6 +59,25 @@ def run_git(*args: str) -> str:
return result.stdout.strip()
def resolve_token(explicit_token: str | None, console: Console) -> str:
"""Resolve Forgejo API token: explicit flag > 1Password."""
if explicit_token:
return explicit_token
console.print("[dim]Reading Forgejo API token from 1Password...[/dim]")
try:
result = subprocess.run(
["op", "read", OP_TOKEN_REF],
capture_output=True,
text=True,
check=True,
)
return result.stdout.strip()
except (subprocess.CalledProcessError, FileNotFoundError) as e:
console.print(f"[red]Failed to read token from 1Password:[/red] {e}")
console.print("[dim]Pass --token explicitly or ensure op CLI is available[/dim]")
raise typer.Exit(1)
def branch_head_age_days(ref: str) -> int | None:
"""Get the age in days of the HEAD commit on a branch ref."""
try:
@ -66,6 +90,15 @@ def branch_head_age_days(ref: str) -> int | None:
return None
def api_branch_age_days(commit_date_str: str) -> int | None:
"""Compute age in days from an ISO date string."""
try:
commit_date = datetime.fromisoformat(commit_date_str)
return (datetime.now(timezone.utc) - commit_date).days
except (ValueError, TypeError):
return None
def get_git_merged_local_branches() -> set[str]:
"""Get local branches that are fully merged into main (fast-forward)."""
try:
@ -110,49 +143,52 @@ def get_all_local_branches() -> set[str]:
return branches
def get_all_remote_branches() -> set[str]:
"""Get all remote branch names (short form, without origin/ prefix)."""
output = run_git("branch", "-r")
branches = set()
for line in output.splitlines():
name = line.strip()
if " -> " in name:
continue
if not name.startswith("origin/"):
continue
short = name.removeprefix("origin/")
if short not in PROTECTED_BRANCHES:
branches.add(short)
def get_api_branches(client: httpx.Client) -> dict[str, str]:
"""Get all remote branches via API. Returns {name: commit_date_iso}."""
branches: dict[str, str] = {}
page = 1
limit = 50
while True:
resp = client.get(
f"{FORGE_API}/repos/{REPO_OWNER}/{REPO_NAME}/branches",
params={"limit": limit, "page": page},
)
resp.raise_for_status()
data = resp.json()
if not data:
break
for branch in data:
name = branch["name"]
if name not in PROTECTED_BRANCHES:
date = branch.get("commit", {}).get("timestamp", "")
branches[name] = date
page += 1
return branches
def get_squash_merged_branches(console: Console) -> set[str]:
"""Query Forgejo API for branch names from squash-merged PRs."""
def get_merged_pr_branches(client: httpx.Client, console: Console) -> set[str]:
"""Query Forgejo API for branch names from merged PRs."""
merged_branches: set[str] = set()
page = 1
limit = 50
try:
with httpx.Client(timeout=15) as client:
while True:
resp = client.get(
f"{FORGE_API}/repos/{REPO}/pulls",
params={"state": "closed", "limit": limit, "page": page},
)
resp.raise_for_status()
prs = resp.json()
if not prs:
break
for pr in prs:
if pr.get("merged"):
ref = pr.get("head", {}).get("ref", "")
if ref and ref not in PROTECTED_BRANCHES:
merged_branches.add(ref)
page += 1
while True:
resp = client.get(
f"{FORGE_API}/repos/{REPO_OWNER}/{REPO_NAME}/pulls",
params={"state": "closed", "limit": limit, "page": page},
)
resp.raise_for_status()
prs = resp.json()
if not prs:
break
for pr in prs:
if pr.get("merged"):
ref = pr.get("head", {}).get("ref", "")
if ref and ref not in PROTECTED_BRANCHES:
merged_branches.add(ref)
page += 1
except httpx.HTTPError as e:
console.print(f"[yellow]Warning:[/yellow] Failed to query Forgejo API: {e}")
console.print("[dim]Falling back to git-only merge detection[/dim]")
console.print(f"[yellow]Warning:[/yellow] Failed to query merged PRs: {e}")
return merged_branches
@ -166,13 +202,16 @@ def delete_local_branch(branch: str) -> tuple[bool, str]:
return False, e.stderr.strip()
def delete_remote_branch(branch: str) -> tuple[bool, str]:
"""Delete a remote branch via git push. Returns (success, message)."""
try:
run_git("push", "origin", "--delete", branch)
def delete_remote_branch_api(
client: httpx.Client, branch: str,
) -> tuple[bool, str]:
"""Delete a remote branch via Forgejo API. Returns (success, message)."""
resp = client.delete(
f"{FORGE_API}/repos/{REPO_OWNER}/{REPO_NAME}/branches/{branch}",
)
if resp.status_code == 204:
return True, "deleted"
except subprocess.CalledProcessError as e:
return False, e.stderr.strip()
return False, f"HTTP {resp.status_code}: {resp.text[:120]}"
app = typer.Typer(add_completion=False)
@ -200,158 +239,175 @@ def main(
bool,
typer.Option("--remote-only", help="Only clean up remote branches"),
] = False,
token: Annotated[
str | None,
typer.Option("--token", help="Forgejo API token (default: read from 1Password)"),
] = None,
) -> None:
"""Delete branches that have been merged into main."""
console = Console()
# Fetch latest remote state
console.print("[dim]Fetching remote branches...[/dim]")
try:
run_git("fetch", "--prune", "origin")
except subprocess.CalledProcessError as e:
console.print(f"[red]Failed to fetch:[/red] {e.stderr}")
raise typer.Exit(1)
# Gather merge info from both sources
console.print("[dim]Checking Forgejo for squash-merged PRs...[/dim]")
api_merged = get_squash_merged_branches(console)
git_merged_local = get_git_merged_local_branches()
git_merged_remote = get_git_merged_remote_branches()
# Union of all confirmed-merged branch names
all_confirmed_merged = api_merged | git_merged_local | git_merged_remote
do_local = not remote_only
do_remote = not local_only
# Remote: delete if confirmed merged by either method
all_remote = get_all_remote_branches() if do_remote else set()
remote_to_delete = sorted(all_remote & all_confirmed_merged)
# Resolve token (needed for API branch listing and deletion)
api_token = resolve_token(token, console) if do_remote else None
# Local: delete if confirmed merged (git --merged OR matched a merged PR)
all_local = get_all_local_branches() if do_local else set()
local_to_delete = sorted(all_local & all_confirmed_merged)
# Compute ages for all candidates
all_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
branch_ages: dict[str, int | None] = {}
for name in all_candidates:
age = None
if name in all_remote:
age = branch_head_age_days(f"origin/{name}")
if age is None and name in all_local:
age = branch_head_age_days(name)
branch_ages[name] = age
# Apply cutoff filter
local_to_delete = [
b for b in local_to_delete
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
]
remote_to_delete = [
b for b in remote_to_delete
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
]
filtered_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
skipped_count = len(all_candidates) - len(filtered_candidates)
if filtered_candidates:
# Build summary table
table = Table(title=f"Merged branches to delete (older than {cutoff} days)")
table.add_column("Branch")
table.add_column("Age (days)", justify="right")
table.add_column("Method")
table.add_column("Local")
table.add_column("Remote")
for branch in filtered_candidates:
has_local = branch in local_to_delete
has_remote = branch in remote_to_delete
age = branch_ages.get(branch)
age_str = str(age) if age is not None else "?"
methods = []
if branch in git_merged_local or branch in git_merged_remote:
methods.append("git")
if branch in api_merged:
methods.append("api")
method_str = "+".join(methods)
table.add_row(
branch,
age_str,
method_str,
"[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]",
"[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]",
)
console.print(table)
console.print(
f"\n[bold]{len(local_to_delete)}[/bold] local, "
f"[bold]{len(remote_to_delete)}[/bold] remote branches to delete"
)
if skipped_count:
console.print(f"[dim]({skipped_count} merged branches skipped — "
f"newer than {cutoff} days)[/dim]")
else:
console.print(f"[green]No merged branches older than "
f"{cutoff} days to clean up.[/green]")
if skipped_count:
console.print(f"[dim]({skipped_count} merged branches skipped — "
f"newer than {cutoff} days)[/dim]")
# Warn about stale unmerged local branches
# Fetch latest remote state for local branch operations
if do_local:
unmerged_local = all_local - all_confirmed_merged
stale_unmerged = []
for name in sorted(unmerged_local):
age = branch_head_age_days(name)
if age is not None and age >= cutoff:
stale_unmerged.append((name, age))
console.print("[dim]Fetching remote branches...[/dim]")
try:
run_git("fetch", "--prune", "origin")
except subprocess.CalledProcessError as e:
console.print(f"[yellow]Warning:[/yellow] git fetch failed: {e.stderr}")
if stale_unmerged:
console.print()
warn_table = Table(
title=f"[yellow]Warning:[/yellow] Stale unmerged local branches "
f"(older than {cutoff} days)",
title_style="",
)
warn_table.add_column("Branch")
warn_table.add_column("Age (days)", justify="right")
for name, age in stale_unmerged:
warn_table.add_row(f"[yellow]{name}[/yellow]", str(age))
console.print(warn_table)
# Gather merge info
console.print("[dim]Checking Forgejo for squash-merged PRs...[/dim]")
with httpx.Client(
timeout=15,
headers={"Authorization": f"token {api_token}"} if api_token else {},
) as client:
api_merged = get_merged_pr_branches(client, console)
git_merged_local = get_git_merged_local_branches() if do_local else set()
git_merged_remote = get_git_merged_remote_branches() if do_local else set()
# Union of all confirmed-merged branch names
all_confirmed_merged = api_merged | git_merged_local | git_merged_remote
# Remote branches and ages via API
remote_branch_ages: dict[str, int | None] = {}
if do_remote:
console.print("[dim]Listing remote branches via API...[/dim]")
api_branches = get_api_branches(client)
for name, date_str in api_branches.items():
remote_branch_ages[name] = api_branch_age_days(date_str)
all_remote = set(api_branches.keys())
else:
all_remote = set()
remote_to_delete = sorted(all_remote & all_confirmed_merged)
# Local branches
all_local = get_all_local_branches() if do_local else set()
local_to_delete = sorted(all_local & all_confirmed_merged)
# Compute ages for all candidates
all_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
branch_ages: dict[str, int | None] = {}
for name in all_candidates:
age = remote_branch_ages.get(name)
if age is None and name in all_local:
age = branch_head_age_days(name)
branch_ages[name] = age
# Apply cutoff filter
local_to_delete = [
b for b in local_to_delete
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
]
remote_to_delete = [
b for b in remote_to_delete
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
]
filtered_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
skipped_count = len(all_candidates) - len(filtered_candidates)
if filtered_candidates:
table = Table(title=f"Merged branches to delete (older than {cutoff} days)")
table.add_column("Branch")
table.add_column("Age (days)", justify="right")
table.add_column("Method")
table.add_column("Local")
table.add_column("Remote")
for branch in filtered_candidates:
has_local = branch in local_to_delete
has_remote = branch in remote_to_delete
age = branch_ages.get(branch)
age_str = str(age) if age is not None else "?"
methods = []
if branch in git_merged_local or branch in git_merged_remote:
methods.append("git")
if branch in api_merged:
methods.append("api")
method_str = "+".join(methods)
table.add_row(
branch,
age_str,
method_str,
"[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]",
"[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]",
)
console.print(table)
console.print(
f"[dim]These {len(stale_unmerged)} branches have no merged PR on "
f"Forgejo and are not git-ancestors of main.\n"
f"They may contain work-in-progress — inspect manually before deleting.[/dim]"
f"\n[bold]{len(local_to_delete)}[/bold] local, "
f"[bold]{len(remote_to_delete)}[/bold] remote branches to delete"
)
if skipped_count:
console.print(f"[dim]({skipped_count} merged branches skipped — "
f"newer than {cutoff} days)[/dim]")
else:
console.print(f"[green]No merged branches older than "
f"{cutoff} days to clean up.[/green]")
if skipped_count:
console.print(f"[dim]({skipped_count} merged branches skipped — "
f"newer than {cutoff} days)[/dim]")
if not filtered_candidates:
raise typer.Exit(0)
# Warn about stale unmerged local branches
if do_local:
unmerged_local = all_local - all_confirmed_merged
stale_unmerged = []
for name in sorted(unmerged_local):
age = branch_head_age_days(name)
if age is not None and age >= cutoff:
stale_unmerged.append((name, age))
if dry_run:
console.print("\n[dim]Dry run — no branches were deleted.[/dim]")
raise typer.Exit(0)
if stale_unmerged:
console.print()
warn_table = Table(
title=f"[yellow]Warning:[/yellow] Stale unmerged local branches "
f"(older than {cutoff} days)",
title_style="",
)
warn_table.add_column("Branch")
warn_table.add_column("Age (days)", justify="right")
for name, age in stale_unmerged:
warn_table.add_row(f"[yellow]{name}[/yellow]", str(age))
console.print(warn_table)
console.print(
f"[dim]These {len(stale_unmerged)} branches have no merged PR on "
f"Forgejo and are not git-ancestors of main.\n"
f"They may contain work-in-progress — inspect manually "
f"before deleting.[/dim]"
)
# Confirm
if not yes and not typer.confirm("\nProceed with deletion?"):
console.print("[dim]Aborted.[/dim]")
raise typer.Exit(0)
if not filtered_candidates:
raise typer.Exit(0)
# Delete remote branches first (source of truth)
if remote_to_delete:
console.print("\n[bold]Deleting remote branches...[/bold]")
for branch in remote_to_delete:
ok, msg = delete_remote_branch(branch)
if ok:
console.print(f" [green]✓[/green] origin/{branch}")
else:
console.print(f" [red]✗[/red] origin/{branch}: {msg}")
if dry_run:
console.print("\n[dim]Dry run — no branches were deleted.[/dim]")
raise typer.Exit(0)
# Delete local branches
# Confirm
if not yes and not typer.confirm("\nProceed with deletion?"):
console.print("[dim]Aborted.[/dim]")
raise typer.Exit(0)
# Delete remote branches via API
if remote_to_delete:
console.print("\n[bold]Deleting remote branches...[/bold]")
for branch in remote_to_delete:
ok, msg = delete_remote_branch_api(client, branch)
if ok:
console.print(f" [green]✓[/green] origin/{branch}")
else:
console.print(f" [red]✗[/red] origin/{branch}: {msg}")
# Delete local branches (outside httpx client context)
if local_to_delete:
console.print("\n[bold]Deleting local branches...[/bold]")
for branch in local_to_delete: