Skip to content

🧹 [pr_auto_rebase.py 리팩토링 - process_queue 의 복잡도 개선] #865

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
seonghobae wants to merge 1 commit into from
Closed

🧹 [pr_auto_rebase.py 리팩토링 - process_queue 의 복잡도 개선] #865

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 31 additions & 26 deletions scripts/ci/pr_auto_rebase.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -545,39 +545,44 @@ def summarize_error(exc: Exception) -> str:
return first_line[:300]



def evaluate_candidate(
args: argparse.Namespace,
pr: dict[str, Any],
now: datetime,
rebases_used: int,
) -> tuple[Decision, int]:
"""Evaluate a single PR candidate and return the decision and updated rate-limit count."""
number = int(pr.get("number") or 0)
skip_reason = candidate_skip_reason(
args.repo,
pr,
base_branch=args.base_branch,
now=now,
human_window_minutes=args.human_window_minutes,
)
if skip_reason is not None:
return Decision(number, "skip", skip_reason), rebases_used
if rebases_used >= args.max_per_run:
return Decision(number, "skip", f"rate limit reached ({args.max_per_run} rebases/run); process next run"), rebases_used
rebases_used += 1
if args.dry_run:
return Decision(number, "would_rebase", f"candidate ({candidate_state_note(pr)}); dry-run, no git mutation"), rebases_used
try:
return perform_rebase(args.repo, pr, dry_run=args.dry_run), rebases_used
except (RuntimeError, KeyError, ValueError) as exc:
return Decision(number, "error", summarize_error(exc)), rebases_used


def process_queue(args: argparse.Namespace) -> int:
"""Inspect open PRs and perform bounded, guarded auto-rebases."""
now = datetime.now(timezone.utc)
prs = fetch_open_prs(args.repo, args.max_prs)
decisions: list[Decision] = []
rebases_used = 0
for pr in prs:
number = int(pr.get("number") or 0)
skip_reason = candidate_skip_reason(
args.repo,
pr,
base_branch=args.base_branch,
now=now,
human_window_minutes=args.human_window_minutes,
)
if skip_reason is not None:
decisions.append(Decision(number, "skip", skip_reason))
continue
if rebases_used >= args.max_per_run:
decisions.append(
Decision(number, "skip", f"rate limit reached ({args.max_per_run} rebases/run); process next run")
)
continue
rebases_used += 1
if args.dry_run:
decisions.append(
Decision(number, "would_rebase", f"candidate ({candidate_state_note(pr)}); dry-run, no git mutation")
)
continue
try:
decisions.append(perform_rebase(args.repo, pr, dry_run=args.dry_run))
except (RuntimeError, KeyError, ValueError) as exc:
decisions.append(Decision(number, "error", summarize_error(exc)))
decision, rebases_used = evaluate_candidate(args, pr, now, rebases_used)
decisions.append(decision)
print_summary(decisions, dry_run=args.dry_run, base_branch=args.base_branch)
return 0

Expand Down
Loading