Fix exceptions. remove copy-paste (#141)

Author: thehighestmath

src/ForgejoRepoAPI.py

@@ -1,5 +1,8 @@
-from github import Github, GithubException
+from github import Github
 
+from src.utils import (
+    log_exceptions,
+)
 from src.interface_wrapper import (
     Branch,
     Comment,
@@ -12,7 +15,6 @@
     Repository,
     User,
     WikiPage,
-    logging,
     WorkflowRun,
 )
 
@@ -22,16 +24,10 @@ def __init__(self, client: Github):
         self.client = self._client_validation(client)
 
     @staticmethod
-    def _client_validation(client: Github) -> Github:
-        try:
-            client.get_user().login
-        except GithubException as err:
-            logging.error(f'Github: Connect: error {err.data}')
-            logging.error(
-                'Github: Connect: user could not be authenticated please try again.'
-            )
-        else:
-            return client
+    @log_exceptions(default_return=None, message="Github: Connect: user could not be authenticated. Please try again.")
+    def _client_validation(client: Github) -> Github | None:
+        client.get_user().login
+        return client
 
     def get_user_data(self, user) -> User:
         return User(
@@ -57,125 +53,96 @@ def get_commit_data(self, commit, files=False) -> Commit:
             deletions=commit.stats.deletions,
         )
 
+    @log_exceptions(default_return=None, message="Failed to get repository from GitHub")
     def get_repository(self, id: str) -> Repository | None:
-        try:
-            repo = self.client.get_repo(id)
-            return Repository(
-                _id=repo.full_name,
-                name=repo.name,
-                url=repo.html_url,
-                default_branch=Branch(name=repo.default_branch, last_commit=None),
-                owner=self.get_user_data(repo.owner),
-            )
-        except Exception as e:
-            logging.error(f"Failed to get repository {id} from GitHub: {e}")
-            return None
+        repo = self.client.get_repo(id)
+        return Repository(
+            _id=repo.full_name,
+            name=repo.name,
+            url=repo.html_url,
+            default_branch=Branch(name=repo.default_branch, last_commit=None),
+            owner=self.get_user_data(repo.owner),
+        )
 
     def get_collaborator_permission(self, repo: Repository, user: User) -> str:
         return self.client.get_repo(repo._id).get_collaborator_permission(user.login)
 
+    @log_exceptions(default_return=[], message="Failed to get commits from GitHub")
     def get_commits(self, repo: Repository, files: bool = True) -> list[Commit]:
-        try:
-            commits = self.client.get_repo(repo._id).get_commits()
-            return [self.get_commit_data(c, files) for c in commits]
-
-        except Exception as e:
-            logging.error(
-                f"Failed to get commits from GitHub for repo {repo.name}: {e}"
-            )
-            return []
+        commits = self.client.get_repo(repo._id).get_commits()
+        return [self.get_commit_data(c, files) for c in commits]
 
+    @log_exceptions(default_return=[], message="Failed to get contributors from GitHub")
     def get_contributors(self, repo: Repository) -> list[Contributor]:
-        try:
-            contributors = self.client.get_repo(repo._id).get_contributors()
-            return [Contributor(c.login, c.email or "") for c in contributors]
-        except Exception as e:
-            logging.error(
-                f"Failed to get contributors from GitHub for repo {repo.name}: {e}"
-            )
-            return []
+        contributors = self.client.get_repo(repo._id).get_contributors()
+        return [Contributor(c.login, c.email or "") for c in contributors]
 
+    @log_exceptions(default_return=[], message="Failed to get issues from GitHub")
     def get_issues(self, repo: Repository) -> list[Issue]:
-        try:
-            issues = self.client.get_repo(repo._id).get_issues(state='all')
-            return [
-                Issue(
-                    _id=i.number,
-                    title=i.title,
-                    state=i.state,
-                    created_at=i.created_at,
-                    closed_at=i.closed_at,
-                    closed_by=self.get_user_data(i.closed_by) if i.closed_by else None,
-                    body=i.body,
-                    user=self.get_user_data(i.user),
-                    labels=[label.name for label in i.labels],
-                    milestone=i.milestone.title if i.milestone else None,
-                )
-                for i in issues
-            ]
-        except Exception as e:
-            logging.error(f"Failed to get issues from GitHub for repo {repo.name}: {e}")
-            return []
+        issues = self.client.get_repo(repo._id).get_issues(state='all')
+        return [
+            Issue(
+                _id=i.number,
+                title=i.title,
+                state=i.state,
+                created_at=i.created_at,
+                closed_at=i.closed_at,
+                closed_by=self.get_user_data(i.closed_by) if i.closed_by else None,
+                body=i.body,
+                user=self.get_user_data(i.user),
+                labels=[label.name for label in i.labels],
+                milestone=i.milestone.title if i.milestone else None,
+            )
+            for i in issues
+        ]
 
+    @log_exceptions(default_return=[], message="Failed to get pull requests from GitHub")
     def get_pull_requests(self, repo: Repository) -> list[PullRequest]:
-        try:
-            pulls = self.client.get_repo(repo._id).get_pulls(state='all')
-            return [
-                PullRequest(
-                    _id=p.number,
-                    title=p.title,
-                    author=self.get_user_data(p.user),
-                    state=p.state,
-                    created_at=p.created_at,
-                    head_label=p.head.label,
-                    base_label=p.base.label,
-                    head_ref=p.head.ref,
-                    base_ref=p.base.ref,
-                    merged_by=self.get_user_data(p.merged_by) if p.merged_by else None,
-                    files=[file.filename for file in p.get_files()],
-                    issue_url=p.issue_url,
-                    labels=[label.name for label in p.labels],
-                    milestone=p.milestone.title if p.milestone else None,
-                )
-                for p in pulls
-            ]
-        except Exception as e:
-            logging.error(
-                f"Failed to get pull requests from GitHub for repo {repo.name}: {e}"
+        pulls = self.client.get_repo(repo._id).get_pulls(state='all')
+        return [
+            PullRequest(
+                _id=p.number,
+                title=p.title,
+                author=self.get_user_data(p.user),
+                state=p.state,
+                created_at=p.created_at,
+                head_label=p.head.label,
+                base_label=p.base.label,
+                head_ref=p.head.ref,
+                base_ref=p.base.ref,
+                merged_by=self.get_user_data(p.merged_by) if p.merged_by else None,
+                files=[file.filename for file in p.get_files()],
+                issue_url=p.issue_url,
+                labels=[label.name for label in p.labels],
+                milestone=p.milestone.title if p.milestone else None,
             )
-            return []
+            for p in pulls
+        ]
 
+    @log_exceptions(default_return=[], message="Failed to get branches from GitHub")
     def get_branches(self, repo: Repository) -> list[Branch]:
-        try:
-            repo_client = self.client.get_repo(repo._id)
-            branches = repo_client.get_branches()
-            result = []
-
-            for branch in branches:
-                commit = repo_client.get_commit(branch.commit.sha)
-
-                commit_obj = self.get_commit_data(commit)
-
-                result.append(Branch(name=branch.name, last_commit=commit_obj))
+        repo_client = self.client.get_repo(repo._id)
+        branches = repo_client.get_branches()
+        result = []
 
-            return result
+        for branch in branches:
+            commit = repo_client.get_commit(branch.commit.sha)
+            commit_obj = self.get_commit_data(commit)
+            result.append(Branch(name=branch.name, last_commit=commit_obj))
 
-        except Exception as e:
-            logging.error(
-                f"Failed to get branches from GitHub for repo {repo.name}: {e}"
-            )
-            return []
+        return result
 
     def get_wiki_pages(self, repo: Repository) -> list[WikiPage]:
-        return
+        raise Exception('not implemented')
 
+    @log_exceptions(default_return=[], message="Failed to get forks from GitHub")
     def get_forks(self, repo: Repository) -> list[Repository]:
         repo_client = self.client.get_repo(repo._id)
         result = []
 
         for r in repo_client.get_forks():
-            default_branch = (Branch(name=r.default_branch, last_commit=None),)
-            owner = (self.get_user_data(r.owner),)
+            default_branch = Branch(name=r.default_branch, last_commit=None)
+            owner = self.get_user_data(r.owner)
 
             result.append(
                 Repository(
@@ -189,83 +156,64 @@ def get_forks(self, repo: Repository) -> list[Repository]:
 
         return result
 
+    @log_exceptions(default_return=[], message="Failed to get comments")
     def get_comments(self, repo, obj) -> list[Comment]:
         repo_client = self.client.get_repo(repo._id)
 
-        try:
-            if isinstance(obj, Issue):
-                # Получаем issue напрямую по номеру
-                issue = repo_client.get_issue(number=obj._id)
-                comments = issue.get_comments()
-            elif isinstance(obj, PullRequest):
-                # Получаем PR напрямую по номеру
-                pull = repo_client.get_pull(number=obj._id)
-                comments = pull.get_comments()
-            else:
-                return []
-
-            # Формируем результат
-            return [
-                Comment(
-                    body=comment.body,
-                    created_at=comment.created_at,
-                    author=self.get_user_data(comment.user),
-                )
-                for comment in comments
-            ]
-
-        except Exception as e:
-            logging.error(f"Failed to get comments for {type(obj).__name__} {obj._id}: {e}")
+        if isinstance(obj, Issue):
+            issue = repo_client.get_issue(number=obj._id)
+            comments = issue.get_comments()
+        elif isinstance(obj, PullRequest):
+            pull = repo_client.get_pull(number=obj._id)
+            comments = pull.get_comments()
+        else:
             return []
 
+        return [
+            Comment(
+                body=comment.body,
+                created_at=comment.created_at,
+                author=self.get_user_data(comment.user),
+            )
+            for comment in comments
+        ]
+
+    @log_exceptions(default_return=[], message="Failed to get invites from GitHub")
     def get_invites(self, repo: Repository) -> list[Invite]:
-        try:
-            invites = self.client.get_repo(repo._id).get_pending_invitations()
-            return [
-                Invite(
-                    _id=i._id,
-                    invitee=self.get_user_data(i.invitee),
-                    created_at=i.created_at,
-                    html_url=i.html_url,
-                )
-                for i in invites
-            ]
-        except Exception as e:
-            logging.error(
-                f"Failed to get invites from GitHub for repo {repo.name}: {e}"
+        invites = self.client.get_repo(repo._id).get_pending_invitations()
+        return [
+            Invite(
+                _id=i._id,
+                invitee=self.get_user_data(i.invitee),
+                created_at=i.created_at,
+                html_url=i.html_url,
             )
-            return []
+            for i in invites
+        ]
 
     def get_rate_limiting(self) -> tuple[int, int]:
         return self.client.rate_limiting
 
+    @log_exceptions(default_return=[], message="Failed to get workflow runs from GitHub")
     def get_workflow_runs(self, repo) -> list[WorkflowRun]:
-        try:
-            runs = self.client.get_repo(repo._id).get_workflow_runs()
-
-            return [
-                WorkflowRun(
-                    display_title=r.display_title,
-                    event=r.event,
-                    head_branch=r.head_branch,
-                    head_sha=r.head_sha,
-                    name=r.name,
-                    path=r.path,
-                    created_at=r.created_at,
-                    run_started_at=r.run_started_at,
-                    updated_at=r.updated_at,
-                    conclusion=r.conclusion,
-                    status=r.status,
-                    url=r.url,
-                )
-                for r in runs
-            ]
-
-        except Exception as e:
-            logging.error(
-                f"Failed to get workflow runs from GitHub for repo {repo.name}: {e}"
+        runs = self.client.get_repo(repo._id).get_workflow_runs()
+        return [
+            WorkflowRun(
+                display_title=r.display_title,
+                event=r.event,
+                head_branch=r.head_branch,
+                head_sha=r.head_sha,
+                name=r.name,
+                path=r.path,
+                created_at=r.created_at,
+                run_started_at=r.run_started_at,
+                updated_at=r.updated_at,
+                conclusion=r.conclusion,
+                status=r.status,
+                url=r.url,
             )
-            return []
+            for r in runs
+        ]
 
     def get_base_url(self) -> str:
         return 'https://api.github.com'

src/GitHubRepoAPI.py

@@ -81,17 +81,13 @@ def log_commits(
     logger.log_to_csv(csv_name, list(info.keys()))
 
     for client, repo, token in binded_repos:
-        try:
-            logger.log_title(repo.name)
-            log_repository_commits(client, repo, csv_name, start, finish, branch)
-            if fork_flag:
-                for forked_repo in client.get_forks(repo):
-                    logger.log_title(f"FORKED: {forked_repo.name}")
-                    log_repository_commits(
-                        client, forked_repo, csv_name, start, finish, branch
-                    )
-                    sleep(TIMEDELTA)
-            sleep(TIMEDELTA)
-        except Exception as e:
-            print(e)
-            exit(1)
+        logger.log_title(repo.name)
+        log_repository_commits(client, repo, csv_name, start, finish, branch)
+        if fork_flag:
+            for forked_repo in client.get_forks(repo):
+                logger.log_title(f"FORKED: {forked_repo.name}")
+                log_repository_commits(
+                    client, forked_repo, csv_name, start, finish, branch
+                )
+                sleep(TIMEDELTA)
+        sleep(TIMEDELTA)