Skip to content
This repository was archived by the owner on Apr 13, 2023. It is now read-only.
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
20 changes: 12 additions & 8 deletions src/git_portfolio/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -182,44 +182,48 @@ def config_repos() -> Union[res.ResponseFailure, res.ResponseSuccess]:


@create.command("issues")
def create_issues() -> None:
@gitp_config_check
def create_issues() -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Batch creation of issues on GitHub."""
manager = ghm.GithubManager(CONFIG_MANAGER.config)
issue = p.InquirerPrompter.create_issues(
CONFIG_MANAGER.config.github_selected_repos
)
ghci.GhCreateIssueUseCase(manager).execute(issue)
return ghci.GhCreateIssueUseCase(manager).execute(issue)


@create.command("prs")
def create_prs() -> None:
@gitp_config_check
def create_prs() -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Batch creation of pull requests on GitHub."""
manager = ghm.GithubManager(CONFIG_MANAGER.config)
pr = p.InquirerPrompter.create_pull_requests(
CONFIG_MANAGER.config.github_selected_repos
)
ghcp.GhCreatePrUseCase(manager).execute(pr)
return ghcp.GhCreatePrUseCase(manager).execute(pr)


@merge.command("prs")
def merge_prs() -> None:
@gitp_config_check
def merge_prs() -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Batch merge of pull requests on GitHub."""
manager = ghm.GithubManager(CONFIG_MANAGER.config)
pr_merge = p.InquirerPrompter.merge_pull_requests(
manager.github_username,
manager.config.github_selected_repos,
)
ghmp.GhMergePrUseCase(manager).execute(pr_merge)
return ghmp.GhMergePrUseCase(manager).execute(pr_merge)


@delete.command("branches")
def delete_branches() -> None:
@gitp_config_check
def delete_branches() -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Batch deletion of branches on GitHub."""
manager = ghm.GithubManager(CONFIG_MANAGER.config)
branch = p.InquirerPrompter.delete_branches(
CONFIG_MANAGER.config.github_selected_repos
)
ghdb.GhDeleteBranchUseCase(manager).execute(branch)
return ghdb.GhDeleteBranchUseCase(manager).execute(branch)


main.add_command(configure)
Expand Down
18 changes: 9 additions & 9 deletions src/git_portfolio/prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@
import inquirer

import git_portfolio.domain.gh_connection_settings as gcs
import git_portfolio.domain.issue as i
import git_portfolio.domain.pull_request as pr
import git_portfolio.domain.pull_request_merge as prm
import git_portfolio.prompt_validation as val
from git_portfolio.domain.issue import Issue
from git_portfolio.domain.pull_request import PullRequest
from git_portfolio.domain.pull_request_merge import PullRequestMerge


class InquirerPrompter:
Expand Down Expand Up @@ -66,7 +66,7 @@ def select_repos(repo_names: List[str]) -> Any:
print("Please select with `space` at least one repo.\n")

@staticmethod
def create_issues(github_selected_repos: List[str]) -> Issue:
def create_issues(github_selected_repos: List[str]) -> i.Issue:
"""Prompt questions to create issues."""
questions = [
inquirer.Text(
Expand All @@ -91,10 +91,10 @@ def create_issues(github_selected_repos: List[str]) -> Issue:
while not correct:
answers = inquirer.prompt(questions)
correct = answers["correct"]
return Issue(answers["title"], answers["body"], answers["labels"])
return i.Issue(answers["title"], answers["body"], answers["labels"])

@staticmethod
def create_pull_requests(github_selected_repos: List[str]) -> PullRequest:
def create_pull_requests(github_selected_repos: List[str]) -> pr.PullRequest:
"""Prompt questions to create pull requests."""
questions = [
inquirer.Text(
Expand Down Expand Up @@ -148,7 +148,7 @@ def create_pull_requests(github_selected_repos: List[str]) -> PullRequest:
while not correct:
answers = inquirer.prompt(questions)
correct = answers["correct"]
return PullRequest(
return pr.PullRequest(
answers["title"],
answers["body"],
answers["labels"],
Expand Down Expand Up @@ -187,7 +187,7 @@ def delete_branches(github_selected_repos: List[str]) -> Any:
@staticmethod
def merge_pull_requests(
github_username: str, github_selected_repos: List[str]
) -> PullRequestMerge:
) -> prm.PullRequestMerge:
"""Prompt questions to merge pull requests."""
questions = [
inquirer.Text(
Expand Down Expand Up @@ -225,7 +225,7 @@ def merge_pull_requests(
while not correct:
answers = inquirer.prompt(questions)
correct = answers["correct"]
return PullRequestMerge(
return prm.PullRequestMerge(
answers["base"],
answers["head"],
answers["prefix"],
Expand Down
29 changes: 17 additions & 12 deletions src/git_portfolio/use_cases/gh_create_issue_use_case.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
"""Create issue on Github use case."""
from typing import Any
from typing import List
from typing import Union

import github

import git_portfolio.domain.issue as i
import git_portfolio.github_manager as ghm
from git_portfolio.domain.issue import Issue
import git_portfolio.response_objects as res


class GhCreateIssueUseCase:
Expand All @@ -15,33 +17,36 @@ def __init__(self, github_manager: ghm.GithubManager) -> None:
"""Initializer."""
self.github_manager = github_manager

def execute(self, issue: Issue, github_repo: str = "") -> None:
def execute(
self, issue: i.Issue, github_repo: str = ""
) -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Create issues."""
labels = (
[label.strip() for label in issue.labels.split(",")] if issue.labels else []
)

if github_repo:
self._create_issue_from_repo(github_repo, issue, labels)
output = self._create_issue_from_repo(github_repo, issue, labels)
else:
output = ""
for github_repo in self.github_manager.config.github_selected_repos:
self._create_issue_from_repo(github_repo, issue, labels)
output += self._create_issue_from_repo(github_repo, issue, labels)
return res.ResponseSuccess(output)

def _create_issue_from_repo(
self, github_repo: str, issue: Any, labels: List[str]
) -> None:
) -> str:
"""Create issue from one repository."""
repo = self.github_manager.github_connection.get_repo(github_repo)
try:
repo.create_issue(title=issue.title, body=issue.body, labels=labels)
print(f"{github_repo}: issue created successfully.")
return f"{github_repo}: issue created successfully."
except github.GithubException as github_exception:
if github_exception.data["message"] == "Issues are disabled for this repo":
print(
(
f"{github_repo}: {github_exception.data['message']}. "
"It may be a fork."
)
return (
f"{github_repo}: {github_exception.data['message']}. "
"It may be a fork."
)

else:
print(f"{github_repo}: {github_exception.data['message']}.")
return f"{github_repo}: {github_exception.data['message']}."
20 changes: 13 additions & 7 deletions src/git_portfolio/use_cases/gh_create_pr_use_case.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
"""Create pull request on Github use case."""
from typing import Any
from typing import Set
from typing import Union

import github

import git_portfolio.domain.pull_request as pr
import git_portfolio.github_manager as ghm
from git_portfolio.domain.pull_request import PullRequest
import git_portfolio.response_objects as res


class GhCreatePrUseCase:
Expand All @@ -15,15 +17,19 @@ def __init__(self, github_manager: ghm.GithubManager) -> None:
"""Initializer."""
self.github_manager = github_manager

def execute(self, pr: PullRequest, github_repo: str = "") -> None:
def execute(
self, pr: pr.PullRequest, github_repo: str = ""
) -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Create pull requests."""
if github_repo:
self._create_pull_request_from_repo(github_repo, pr)
output = self._create_pull_request_from_repo(github_repo, pr)
else:
output = ""
for github_repo in self.github_manager.config.github_selected_repos:
self._create_pull_request_from_repo(github_repo, pr)
output += self._create_pull_request_from_repo(github_repo, pr)
return res.ResponseSuccess(output)

def _create_pull_request_from_repo(self, github_repo: str, pr: Any) -> None:
def _create_pull_request_from_repo(self, github_repo: str, pr: Any) -> str:
"""Create pull request from one repository."""
repo = self.github_manager.github_connection.get_repo(github_repo)
body = pr.body
Expand All @@ -40,18 +46,18 @@ def _create_pull_request_from_repo(self, github_repo: str, pr: Any) -> None:
base=pr.base,
draft=pr.draft,
)
print(f"{github_repo}: PR created successfully.")
# PyGithub does not support a list of strings for adding (only one str)
for label in labels:
created_pr.add_to_labels(label)
return f"{github_repo}: PR created successfully."
except github.GithubException as github_exception:
extra = ""
for error in github_exception.data["errors"]:
if "message" in error:
extra += f"{error['message']} " # type: ignore
else:
extra += f"Invalid field {error['field']}. " # type: ignore
print(f"{github_repo}: {github_exception.data['message']}. {extra}")
return f"{github_repo}: {github_exception.data['message']}. {extra}"

@staticmethod
def _link_issues(body: str, labels: Set[Any], pr: Any, repo: Any) -> Any:
Expand Down
19 changes: 13 additions & 6 deletions src/git_portfolio/use_cases/gh_delete_branch_use_case.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
"""Delete branch on Github use case."""
from typing import Union

import github

import git_portfolio.github_manager as ghm
import git_portfolio.response_objects as res


class GhDeleteBranchUseCase:
Expand All @@ -11,20 +14,24 @@ def __init__(self, github_manager: ghm.GithubManager) -> None:
"""Initializer."""
self.github_manager = github_manager

def execute(self, branch: str, github_repo: str = "") -> None:
def execute(
self, branch: str, github_repo: str = ""
) -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Delete branches."""
if github_repo:
self._delete_branch_from_repo(github_repo, branch)
output = self._delete_branch_from_repo(github_repo, branch)
else:
output = ""
for github_repo in self.github_manager.config.github_selected_repos:
self._delete_branch_from_repo(github_repo, branch)
output += self._delete_branch_from_repo(github_repo, branch)
return res.ResponseSuccess(output)

def _delete_branch_from_repo(self, github_repo: str, branch: str) -> None:
def _delete_branch_from_repo(self, github_repo: str, branch: str) -> str:
"""Delete a branch from one repository."""
repo = self.github_manager.github_connection.get_repo(github_repo)
try:
git_ref = repo.get_git_ref(f"heads/{branch}")
git_ref.delete()
print(f"{github_repo}: branch deleted successfully.")
return f"{github_repo}: branch deleted successfully."
except github.GithubException as github_exception:
print(f"{github_repo}: {github_exception.data['message']}.")
return f"{github_repo}: {github_exception.data['message']}."
41 changes: 23 additions & 18 deletions src/git_portfolio/use_cases/gh_merge_pr_use_case.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
"""Merge pull request on Github use case."""
from typing import Any
from typing import Union

import github

import git_portfolio.domain.pull_request_merge as prm
import git_portfolio.github_manager as ghm
import git_portfolio.response_objects as res
import git_portfolio.use_cases.gh_delete_branch_use_case as dbr
from git_portfolio.domain.pull_request_merge import PullRequestMerge


class GhMergePrUseCase:
Expand All @@ -15,7 +17,9 @@ def __init__(self, github_manager: ghm.GithubManager) -> None:
"""Initializer."""
self.github_manager = github_manager

def execute(self, pr_merge: PullRequestMerge, github_repo: str = "") -> None:
def execute(
self, pr_merge: prm.PullRequestMerge, github_repo: str = ""
) -> Union[res.ResponseFailure, res.ResponseSuccess]:
"""Merge pull requests."""
# Important note: base and head arguments have different import formats.
# https://developer.github.com/v3/pulls/#list-pull-requests
Expand All @@ -24,41 +28,42 @@ def execute(self, pr_merge: PullRequestMerge, github_repo: str = "") -> None:
state = "open"

if github_repo:
self._merge_pull_request_from_repo(github_repo, head, pr_merge, state)
output = self._merge_pull_request_from_repo(
github_repo, head, pr_merge, state
)
else:
output = ""
for github_repo in self.github_manager.config.github_selected_repos:
self._merge_pull_request_from_repo(github_repo, head, pr_merge, state)
output += self._merge_pull_request_from_repo(
github_repo, head, pr_merge, state
)
return res.ResponseSuccess(output)

def _merge_pull_request_from_repo(
self, github_repo: str, head: str, pr_merge: Any, state: str
) -> None:
) -> str:
"""Merge pull request from one repository."""
repo = self.github_manager.github_connection.get_repo(github_repo)
pulls = repo.get_pulls(state=state, base=pr_merge.base, head=head)
if pulls.totalCount == 1:
pull = pulls[0]
if pull.mergeable:
output = ""
try:
pull.merge()
print(f"{github_repo}: PR merged successfully.")
output += f"{github_repo}: PR merged successfully.\n"
if pr_merge.delete_branch:
delete_branch_use_case = dbr.GhDeleteBranchUseCase(
self.github_manager
)
delete_branch_use_case.execute(pr_merge.head, github_repo)
except github.GithubException as github_exception:
print(f"{github_repo}: {github_exception.data['message']}.")
output += f"{github_repo}: {github_exception.data['message']}."
return output
else:
print(
(
f"{github_repo}: PR not mergeable, GitHub checks may be "
"running."
)
)
return f"{github_repo}: PR not mergeable, GitHub checks may be running."
else:
print(
(
f"{github_repo}: no open PR found for {pr_merge.base}:"
f"{pr_merge.head}."
)
return (
f"{github_repo}: no open PR found for {pr_merge.base}:"
f"{pr_merge.head}."
)
Loading