Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
96 changes: 44 additions & 52 deletions jeeves_pr_stack/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from rich.console import Console
from rich.prompt import Confirm, Prompt
from rich.style import Style
from sh import gh, git
from sh import git
from typer import Argument, Exit, Option, Typer

from jeeves_pr_stack import github
Expand All @@ -18,8 +18,8 @@
from jeeves_pr_stack.models import PRStackContext, PullRequest, State

app = Typer(
help='Manage stacks of GitHub PRs.',
name='stack',
help="Manage stacks of GitHub PRs.",
name="stack",
invoke_without_command=True,
)

Expand Down Expand Up @@ -56,51 +56,51 @@ def print_current_stack(context: PRStackContext):
return

console.print(
'∅ No PRs associated with current branch.\n',
style=Style(color='white', bold=True),
"∅ No PRs associated with current branch.\n",
style=Style(color="white", bold=True),
)
console.print('• Use [code]gh pr create[/code] to create one,')
console.print("• Use [code]gh pr create[/code] to create one,")
console.print(
'• Or [code]j stack push[/code] to stack it onto another PR.',
"• Or [code]j stack push[/code] to stack it onto another PR.",
)
console.print()
console.print('Get more help with [code]j stack --help[/code].')
console.print("Get more help with [code]j stack --help[/code].")


@app.command()
def pop(context: PRStackContext): # noqa: WPS213
"""Merge the bottom-most PR of current stack to the main branch."""
if not context.obj.stack:
raise ValueError('Nothing to merge, current stack is empty.')
raise ValueError("Nothing to merge, current stack is empty.")

top_pr, *remaining_prs = context.obj.stack

default_branch = github.retrieve_default_branch()
console = Console()
if top_pr.base_branch != default_branch:
raise ValueError('Base branch of the PR ≠ default branch of the repo.')
raise ValueError("Base branch of the PR ≠ default branch of the repo.")

dependant_pr = None
if remaining_prs:
dependant_pr = funcy.first(remaining_prs)

console.print('PR to merge: ', top_pr)
console.print('Dependant PR: ', dependant_pr)
console.print("PR to merge: ", top_pr)
console.print("Dependant PR: ", dependant_pr)

if not Confirm.ask('Do you confirm?', default=True):
console.print('Aborted.', style='red')
if not Confirm.ask("Do you confirm?", default=True):
console.print("Aborted.", style="red")
raise Exit(1)

if dependant_pr is not None:
console.print(f'Changing base of {dependant_pr} to {default_branch}')
github.update_pr_base(gh, dependant_pr.number, default_branch)
console.print(f"Changing base of {dependant_pr} to {default_branch}")
github.update_pr_base(context.obj.gh, dependant_pr.number, default_branch)

console.print(f'Merging {top_pr}...')
gh.pr.merge('--merge', top_pr.number)
console.print(f"Merging {top_pr}...")
context.obj.gh.pr.merge("--merge", top_pr.number)

console.print(f'Deleting branch: {top_pr.branch}')
git.push.origin('--delete', top_pr.branch)
console.print('OK.')
console.print(f"Deleting branch: {top_pr.branch}")
git.push.origin("--delete", top_pr.branch)
console.print("OK.")


@app.command()
Expand All @@ -116,7 +116,7 @@ def _ask_for_pull_request_number(pull_requests: list[PullRequest]) -> int:

return int(
Prompt.ask(
'Select the PR',
"Select the PR",
choices=choices,
show_choices=True,
default=funcy.first(choices),
Expand All @@ -126,18 +126,11 @@ def _ask_for_pull_request_number(pull_requests: list[PullRequest]) -> int:

def filter_appendable(pull_requests: list[PullRequest]) -> list[PullRequest]:
"""Determine which PRs we can direct a new PR to."""
directed_to = {
pr.base_branch: pr.branch
for pr in pull_requests
}
directed_to = {pr.base_branch: pr.branch for pr in pull_requests}

return sorted(
[
pr
for pr in pull_requests
if directed_to.get(pr.branch) is None
],
key=operator.attrgetter('number'),
[pr for pr in pull_requests if directed_to.get(pr.branch) is None],
key=operator.attrgetter("number"),
reverse=True,
)

Expand All @@ -146,8 +139,8 @@ def filter_appendable(pull_requests: list[PullRequest]) -> list[PullRequest]:
def push(
author: Annotated[
str,
Option(help='Author of the PRs to choose from.'),
] = '@me',
Option(help="Author of the PRs to choose from."),
] = "@me",
pull_request_id: Annotated[Optional[int], Argument()] = None,
):
"""Direct current branch/PR to an existing PR."""
Expand All @@ -159,22 +152,21 @@ def push(
application.list_pull_requests(author=author),
)
if not pull_requests_to_append:
raise ValueError('No PRs found which this branch could refer to.')
raise ValueError("No PRs found which this branch could refer to.")

console.print(f'Current branch:\n {application.starting_branch}\n')
console.print(f"Current branch:\n {application.starting_branch}\n")

if pull_request_id is None:
pull_request_id = _ask_for_pull_request_number(pull_requests_to_append)

base_pull_request = {
pr.number: pr
for pr in pull_requests_to_append
}[pull_request_id]
base_pull_request = {pr.number: pr for pr in pull_requests_to_append}[
pull_request_id
]

# FIXME: Handle update of existing PR instead of creating a new one
application.gh.pr.create(
base=base_pull_request.branch,
assignee='@me',
assignee="@me",
_fg=True,
)

Expand All @@ -187,20 +179,20 @@ def rebase(context: PRStackContext):
console = Console()
for is_not_first, pr in enumerate(application.rebase()):
if is_not_first:
console.print('OK', style='green')
console.print("OK", style="green")

console.print(
f'#{pr.number} {pr.title}… ',
end='',
f"#{pr.number} {pr.title}… ",
end="",
)

console.print('OK', style='green')
console.print("OK", style="green")
console.print()
console.print('✔ Stack 🥞 rebased.', style='green')
console.print("✔ Stack 🥞 rebased.", style="green")


@app.command()
def split(context: PRStackContext): # noqa: WPS210
def split(context: PRStackContext): # noqa: WPS210
"""Split current PR by commit."""
application = JeevesPullRequestStack()

Expand All @@ -213,23 +205,23 @@ def split(context: PRStackContext): # noqa: WPS210
if original_pull_request is None:
raise NoPullRequestOnBranch(branch=context.obj.current_branch)

console.print('Commits:')
console.print("Commits:")
for commit_number, commit in enumerated_commits:
console.print(f'#{commit_number} {commit.title}')
console.print(f"#{commit_number} {commit.title}")

choices = [str(number) for number, _commit in enumerated_commits]
splitting_commit_number = Prompt.ask(
prompt='Please choose the commit by which to split the PR',
prompt="Please choose the commit by which to split the PR",
default=funcy.first(choices),
choices=choices,
show_choices=True,
)

splitting_commit = dict(enumerated_commits)[int(splitting_commit_number)]

console.print('Current branch:')
console.print("Current branch:")
console.print(context.obj.current_branch)
new_branch_name = Prompt.ask(prompt='Enter the new branch name')
new_branch_name = Prompt.ask(prompt="Enter the new branch name")

application.split(
splitting_commit=splitting_commit,
Expand Down
55 changes: 26 additions & 29 deletions jeeves_pr_stack/github.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,50 +12,47 @@ def construct_checks_status(raw_pull_request: RawPullRequest) -> ChecksStatus:
"""Analyze checks for PR and express their status as one value."""
raw_status_values = {
conclusion
for check in raw_pull_request['statusCheckRollup']
if (conclusion := check.get('conclusion'))
for check in raw_pull_request["statusCheckRollup"]
if (conclusion := check.get("conclusion"))
}

# This one is not informative
raw_status_values.discard('SUCCESS')
raw_status_values.discard("SUCCESS")

# No idea what to do with this one
raw_status_values.discard('NEUTRAL')
raw_status_values.discard("NEUTRAL")

# We do not care
raw_status_values.discard('SKIPPED')
raw_status_values.discard('CANCELLED')
raw_status_values.discard("SKIPPED")
raw_status_values.discard("CANCELLED")

try:
raw_status_values.remove('')
raw_status_values.remove("")
except KeyError:
pass # noqa: WPS420
pass # noqa: WPS420
else:
return ChecksStatus.RUNNING

try:
raw_status_values.remove('FAILURE')
raw_status_values.remove("FAILURE")
except KeyError:
# No failures detected, we are fine
pass # noqa: WPS420
pass # noqa: WPS420
else:
return ChecksStatus.FAILURE

if raw_status_values:
raise ValueError(f'Unknown check statuses: {raw_status_values}')
raise ValueError(f"Unknown check statuses: {raw_status_values}")

return ChecksStatus.SUCCESS


def construct_stack_for_branch( # noqa: WPS210
def construct_stack_for_branch( # noqa: WPS210
branch: str,
pull_requests: list[PullRequest],
) -> list[PullRequest]:
"""Construct sequence of PRs that covers the given branch."""
pull_request_by_branch = {
pr.branch: pr
for pr in pull_requests
}
pull_request_by_branch = {pr.branch: pr for pr in pull_requests}

graph = DiGraph(
incoming_graph_data=[
Expand All @@ -67,37 +64,36 @@ def construct_stack_for_branch( # noqa: WPS210

successors = [
(source, destination)
for source, destination, _reverse # noqa: WPS361
in edge_dfs(graph, source=branch, orientation='reverse')
for source, destination, _reverse in edge_dfs( # noqa: WPS361
graph, source=branch, orientation="reverse"
)
]
predecessors = list(reversed(list(edge_dfs(graph, source=branch))))
edges = predecessors + successors

return [
pull_request_by_branch[branch]
for branch, _base_branch in edges
]
return [pull_request_by_branch[branch] for branch, _base_branch in edges]


def retrieve_current_branch() -> str:
"""Retrieve current git branch name."""
return git.branch('--show-current').strip()
return git.branch("--show-current").strip()


def _construct_gh_env() -> dict[str, str]:
return {
**os.environ,
'NO_COLOR': '1',
"NO_COLOR": "1",
}


def construct_gh_command() -> Command:
"""Construct the GitHub CLI command."""
return gh.bake(
_long_sep=None,
_tty_out=False,
_env={
**os.environ,
'NO_COLOR': '1',
"NO_COLOR": "1",
},
)

Expand All @@ -106,10 +102,11 @@ def retrieve_default_branch() -> str:
"""Get default branch of current repository."""
return json.loads(
gh.repo.view(
json='defaultBranchRef',
json="defaultBranchRef",
_env=_construct_gh_env(),
_tty_out=False,
),
)['defaultBranchRef']['name']
)["defaultBranchRef"]["name"]


def update_pr_base(gh_cmd, pr_number: int, base_branch: str) -> None:
Expand All @@ -122,8 +119,8 @@ def update_pr_base(gh_cmd, pr_number: int, base_branch: str) -> None:
stderr = stderr_raw.decode()
else:
stderr = str(stderr_raw)
projects_classic = 'Projects (classic)' in stderr
project_cards = 'repository.pullRequest.projectCards' in stderr
projects_classic = "Projects (classic)" in stderr
project_cards = "repository.pullRequest.projectCards" in stderr
if projects_classic and project_cards:
raise GhPrEditDeprecationError() from err
raise
Loading