Migrate update_reviewers to subcommand

ryujinx_mako/commands/_subcommand.py

@@ -0,0 +1,44 @@
+import logging
+from abc import ABC, abstractmethod
+from argparse import ArgumentParser, Namespace
+
+from github import Github
+from github.Auth import AppAuth
+
+from ryujinx_mako._const import APP_ID, PRIVATE_KEY, INSTALLATION_ID, SCRIPT_NAME
+
+
+class Subcommand(ABC):
+    @abstractmethod
+    def __init__(self, parser: ArgumentParser):
+        parser.set_defaults(func=self.run)
+
+    @property
+    def logger(self):
+        return logging.getLogger(SCRIPT_NAME).getChild(
+            type(self).name().replace("-", "_")
+        )
+
+    @abstractmethod
+    def run(self, args: Namespace):
+        raise NotImplementedError()
+
+    @staticmethod
+    @abstractmethod
+    def name() -> str:
+        raise NotImplementedError()
+
+    @staticmethod
+    @abstractmethod
+    def description() -> str:
+        raise NotImplementedError()
+
+
+class GithubSubcommand(Subcommand, ABC):
+    _github = Github(
+        auth=AppAuth(APP_ID, PRIVATE_KEY).get_installation_auth(INSTALLATION_ID)
+    )
+
+    @property
+    def github(self):
+        return type(self)._github

ryujinx_mako/commands/update_reviewers.py

@@ -0,0 +1,87 @@
+from argparse import ArgumentParser, Namespace
+from pathlib import Path
+from github.Repository import Repository
+from github.GithubException import GithubException
+
+import yaml
+
+from ryujinx_mako.commands._subcommand import GithubSubcommand
+
+
+class UpdateReviewers(GithubSubcommand):
+    @staticmethod
+    def name() -> str:
+        return "update-reviewers"
+
+    @staticmethod
+    def description() -> str:
+        return "Update reviewers for the specified PR"
+
+    def __init__(self, parser: ArgumentParser):
+        self._reviewers = set()
+        self._team_reviewers = set()
+
+        parser.add_argument("repo_path", type=str)
+        parser.add_argument("pr_number", type=int)
+        parser.add_argument("config_path", type=Path)
+
+        super().__init__(parser)
+
+    def add_reviewers(self, new_entries: list[str]):
+        for reviewer in new_entries:
+            if reviewer.startswith("@"):
+                self._team_reviewers.add(reviewer[1:])
+            else:
+                self._reviewers.add(reviewer)
+
+    def update_reviewers(self, config, repo: Repository, pr_number: int) -> int:
+        pull_request = repo.get_pull(pr_number)
+
+        if not pull_request:
+            self.logger.error(f"Unknown PR #{pr_number}")
+            return 1
+
+        if pull_request.draft:
+            self.logger.warning("Not assigning reviewers for draft PRs")
+            return 0
+
+        pull_request_author = pull_request.user.login
+
+        for label in pull_request.labels:
+            if label.name in config:
+                self.add_reviewers(config[label.name])
+
+        if "default" in config:
+            self.add_reviewers(config["default"])
+
+        if pull_request_author in self._reviewers:
+            self._reviewers.remove(pull_request_author)
+
+        try:
+            reviewers = list(self._reviewers)
+            team_reviewers = list(self._team_reviewers)
+            self.logger.info(
+                f"Attempting to assign reviewers ({reviewers}) "
+                f"and team_reviewers ({team_reviewers})"
+            )
+            pull_request.create_review_request(reviewers, team_reviewers)
+            return 0
+        except GithubException:
+            self.logger.exception(f"Cannot assign review request for PR #{pr_number}")
+            return 1
+
+    def run(self, args: Namespace):
+        repo = self.github.get_repo(args.repo_path)
+
+        if not repo:
+            self.logger.error("Repository not found!")
+            exit(1)
+
+        if not args.config_path.exists():
+            self.logger.error(f"Config '{args.config_path}' not found!")
+            exit(1)
+
+        with open(args.config_path, "r") as file:
+            config = yaml.safe_load(file)
+
+        exit(self.update_reviewers(config, repo, args.pr_number))

ryujinx_mako/update_reviewers.py

@@ -1,87 +0,0 @@
-from pathlib import Path
-from typing import List, Set
-from github import Auth, Github
-from github.Repository import Repository
-from github.GithubException import GithubException
-
-import os
-import sys
-import yaml
-
-
-def add_reviewers(
-    reviewers: Set[str], team_reviewers: Set[str], new_entries: List[str]
-):
-    for reviewer in new_entries:
-        if reviewer.startswith("@"):
-            team_reviewers.add(reviewer[1:])
-        else:
-            reviewers.add(reviewer)
-
-
-def update_reviewers(config, repo: Repository, pr_id: int) -> int:
-    pull_request = repo.get_pull(pr_id)
-
-    if not pull_request:
-        sys.stderr.write(f"Unknown PR #{pr_id}\n")
-        return 1
-
-    if pull_request.draft:
-        print("Not assigning reviewers for draft PRs")
-        return 0
-
-    pull_request_author = pull_request.user.login
-    reviewers = set()
-    team_reviewers = set()
-
-    for label in pull_request.labels:
-        if label.name in config:
-            add_reviewers(reviewers, team_reviewers, config[label.name])
-
-    if "default" in config:
-        add_reviewers(reviewers, team_reviewers, config["default"])
-
-    if pull_request_author in reviewers:
-        reviewers.remove(pull_request_author)
-
-    try:
-        reviewers = list(reviewers)
-        team_reviewers = list(team_reviewers)
-        print(
-            f"Attempting to assign reviewers ({reviewers}) and team_reviewers ({team_reviewers})"
-        )
-        pull_request.create_review_request(reviewers, team_reviewers)
-        return 0
-    except GithubException as e:
-        sys.stderr.write(f"Cannot assign review request for PR #{pr_id}: {e}\n")
-        return 1
-
-
-if __name__ == "__main__":
-    if len(sys.argv) != 7:
-        sys.stderr.write("usage: <app_id> <private_key_env_name> <installation_id> <repo_path> <pr_id> <config_path>\n")
-        sys.exit(1)
-
-    app_id = int(sys.argv[1])
-    private_key = os.environ[sys.argv[2]]
-    installation_id = int(sys.argv[3])
-    repo_path = sys.argv[4]
-    pr_id = int(sys.argv[5])
-    config_path = Path(sys.argv[6])
-
-    auth = Auth.AppAuth(app_id, private_key).get_installation_auth(installation_id)
-    g = Github(auth=auth)
-    repo = g.get_repo(repo_path)
-
-    if not repo:
-        sys.stderr.write("Repository not found!\n")
-        sys.exit(1)
-
-    if not config_path.exists():
-        sys.stderr.write(f'Config "{config_path}" not found!\n')
-        sys.exit(1)
-
-    with open(config_path, "r") as f:
-        config = yaml.safe_load(f)
-
-    sys.exit(update_reviewers(config, repo, pr_id))