apply black

Author: thehighestmath

ForgejoRepoAPI.py

@@ -10,7 +10,7 @@
     PullRequest,
     WikiPage,
     Comment,
-    Invite
+    Invite,
 )
 import base64
 import sys
@@ -32,12 +32,14 @@ def get_user_data(self, user) -> User:
             type=getattr(user, 'type', ""),
             bio=getattr(user, 'bio', ""),
             site_admin=user.site_admin if hasattr(user, 'site_admin') else False,
-            _id=user.id
+            _id=user.id,
         )
 
     def get_repository(self, id: str) -> Repository | None:
         try:
-            repo = self.client.repository.repo_get(owner=id.split('/')[0], repo=id.split('/')[1])
+            repo = self.client.repository.repo_get(
+                owner=id.split('/')[0], repo=id.split('/')[1]
+            )
 
             if not repo:
                 logging.error(f"Failed to get repository {id} from Forgejo.")
@@ -57,30 +59,35 @@ def get_repository(self, id: str) -> Repository | None:
     def get_collaborator_permission(self, repo: Repository, user: User) -> str:
         try:
             permission = self.client.repository.repo_get_repo_permissions(
-                owner=repo.owner.login,
-                repo=repo.name,
-                collaborator=user.login
+                owner=repo.owner.login, repo=repo.name, collaborator=user.login
             )
             return permission.permission
 
         except Exception as e:
             if ("401" in str(e)) or ("403" in str(e)):
                 logging.error(
-                    f"Permission error: Only admins or repo admins can view permissions for others in {repo.name}.")
+                    f"Permission error: Only admins or repo admins can view permissions for others in {repo.name}."
+                )
                 return f"Permission error: Only admins or repo admins can view permissions for others in {repo.name}."
-            logging.error(f"Failed to get collaborator permission for {user.login} in {repo.name}: {e}")
+            logging.error(
+                f"Failed to get collaborator permission for {user.login} in {repo.name}: {e}"
+            )
             return "Error"
 
     def get_commits(self, repo: Repository, files: bool = True) -> list[Commit]:
         try:
-            commits = self.client.repository.repo_get_all_commits(repo.owner.login, repo.name)
+            commits = self.client.repository.repo_get_all_commits(
+                repo.owner.login, repo.name
+            )
             return [
                 Commit(
                     _id=c.sha,
                     message=c.commit.message,
                     author=self.get_user_data(c.author),
                     date=isodate.parse_datetime(c.commit.author.date),
-                    files=[f.filename for f in getattr(c, "files", [])] if files else None
+                    files=(
+                        [f.filename for f in getattr(c, "files", [])] if files else None
+                    ),
                 )
                 for c in commits
             ]
@@ -92,11 +99,17 @@ def get_commits(self, repo: Repository, files: bool = True) -> list[Commit]:
 
     def get_contributors(self, repo: Repository) -> list[Contributor]:
         try:
-            commits = self.client.repository.repo_get_all_commits(repo.owner.login, repo.name)
-            contributors = {c.author.login: c.author.email or "" for c in commits if c.author}
+            commits = self.client.repository.repo_get_all_commits(
+                repo.owner.login, repo.name
+            )
+            contributors = {
+                c.author.login: c.author.email or "" for c in commits if c.author
+            }
             return [Contributor(login, email) for login, email in contributors.items()]
         except Exception as e:
-            logging.error(f"Failed to get contributors from Forgejo for repo {repo.name}: {e}")
+            logging.error(
+                f"Failed to get contributors from Forgejo for repo {repo.name}: {e}"
+            )
             return []
 
     def get_issues(self, repo: Repository) -> list[Issue]:
@@ -109,7 +122,11 @@ def get_issues(self, repo: Repository) -> list[Issue]:
                     state=i.state,
                     created_at=i.created_at,
                     closed_at=i.closed_at if i.state == 'closed' else None,
-                    closed_by=self.get_user_data(i.closed_by) if hasattr(i, 'closed_by') and i.closed_by else None,
+                    closed_by=(
+                        self.get_user_data(i.closed_by)
+                        if hasattr(i, 'closed_by') and i.closed_by
+                        else None
+                    ),
                     body=i.body,
                     user=self.get_user_data(i.user),
                     labels=[label.name for label in i.labels],
@@ -118,12 +135,16 @@ def get_issues(self, repo: Repository) -> list[Issue]:
                 for i in issues
             ]
         except Exception as e:
-            logging.error(f"Failed to get issues from Forgejo for repo {repo.name}: {e}")
+            logging.error(
+                f"Failed to get issues from Forgejo for repo {repo.name}: {e}"
+            )
             return []
 
     def get_pull_requests(self, repo: Repository) -> list[PullRequest]:
         try:
-            pulls = self.client.repository.repo_list_pull_requests(repo.owner.login, repo.name)
+            pulls = self.client.repository.repo_list_pull_requests(
+                repo.owner.login, repo.name
+            )
             return [
                 PullRequest(
                     _id=p.number,
@@ -144,12 +165,16 @@ def get_pull_requests(self, repo: Repository) -> list[PullRequest]:
                 for p in pulls
             ]
         except Exception as e:
-            logging.error(f"Failed to get pull requests from Forgejo for repo {repo.name}: {e}")
+            logging.error(
+                f"Failed to get pull requests from Forgejo for repo {repo.name}: {e}"
+            )
             return []
 
     def get_branches(self, repo: Repository) -> list[Branch]:
         try:
-            branches = self.client.repository.repo_list_branches(repo.owner.login, repo.name)
+            branches = self.client.repository.repo_list_branches(
+                repo.owner.login, repo.name
+            )
             result = []
 
             for branch in branches:
@@ -161,7 +186,9 @@ def get_branches(self, repo: Repository) -> list[Branch]:
                     email=author.email if author and author.email else "",
                 )
 
-                commit_details = self.client.repository.repo_get_single_commit(repo.owner.login, repo.name, commit.id)
+                commit_details = self.client.repository.repo_get_single_commit(
+                    repo.owner.login, repo.name, commit.id
+                )
                 files = [file.filename for file in getattr(commit_details, "files", [])]
 
                 commit_obj = Commit(
@@ -177,27 +204,37 @@ def get_branches(self, repo: Repository) -> list[Branch]:
             return result
 
         except Exception as e:
-            logging.error(f"Failed to get branches from Forgejo for repo {repo.name}: {e}")
+            logging.error(
+                f"Failed to get branches from Forgejo for repo {repo.name}: {e}"
+            )
             return []
 
     def get_wiki_pages(self, repo: Repository) -> list[WikiPage]:
         try:
-            pages = self.client.repository.repo_get_wiki_pages(repo.owner.login, repo.name)
+            pages = self.client.repository.repo_get_wiki_pages(
+                repo.owner.login, repo.name
+            )
             result = []
 
             for page in pages:
-                page_details = self.client.repository.repo_get_wiki_page(repo.owner.login, repo.name, page.title)
+                page_details = self.client.repository.repo_get_wiki_page(
+                    repo.owner.login, repo.name, page.title
+                )
 
                 wiki_page = WikiPage(
                     title=page_details.title,
-                    content=base64.b64decode(page_details.content_base_64).decode('utf-8')
+                    content=base64.b64decode(page_details.content_base_64).decode(
+                        'utf-8'
+                    ),
                 )
                 result.append(wiki_page)
 
             return result
 
         except Exception as e:
-            logging.error(f"Failed to get wiki pages from Forgejo for repo {repo.name}: {e}")
+            logging.error(
+                f"Failed to get wiki pages from Forgejo for repo {repo.name}: {e}"
+            )
             return []
 
     def get_forks(self, repo: Repository) -> list[Repository]:
@@ -215,8 +252,7 @@ def get_forks(self, repo: Repository) -> list[Repository]:
                         name=fork.name,
                         url=fork.html_url,
                         default_branch=default_branch,
-                        owner=owner
-
+                        owner=owner,
                     )
                 )
             return result
@@ -229,7 +265,9 @@ def get_comments(self, repo, obj) -> list[Comment]:
         result = []
         try:
             if isinstance(obj, Issue):
-                comments = self.client.issue.get_repo_comments(repo.owner.login, repo.name)
+                comments = self.client.issue.get_repo_comments(
+                    repo.owner.login, repo.name
+                )
                 result = [
                     Comment(
                         body=c.body,
@@ -240,8 +278,9 @@ def get_comments(self, repo, obj) -> list[Comment]:
                 ]
 
             elif isinstance(obj, PullRequest):
-                comments = self.client.repository.repo_get_pull_review_comments(repo.owner.login, repo.name, obj._id,
-                                                                                100000)  # нет id комментария
+                comments = self.client.repository.repo_get_pull_review_comments(
+                    repo.owner.login, repo.name, obj._id, 100000
+                )  # нет id комментария
                 result = [
                     Comment(
                         body=c.body,

git_logger.py

@@ -1,7 +1,4 @@
-from interface_wrapper import (
-    RepositoryFactory,
-    IRepositoryAPI
-)
+from interface_wrapper import RepositoryFactory, IRepositoryAPI
 from time import sleep
 
 TIMEDELTA = 0.05
@@ -25,8 +22,13 @@ def __init__(self, source: str, tokens: list[str], base_url: str | None = None):
         self.clients = self._init_clients(source, tokens, base_url)
         self.cur_client = None
 
-    def _init_clients(self, source: str, tokens: list[str], base_url: str | None) -> list[dict]:
-        clients = [{"client": login(source, token, base_url), "token": token} for token in tokens]
+    def _init_clients(
+        self, source: str, tokens: list[str], base_url: str | None
+    ) -> list[dict]:
+        clients = [
+            {"client": login(source, token, base_url), "token": token}
+            for token in tokens
+        ]
 
         return clients
 

interface_wrapper.py

@@ -175,15 +175,19 @@ def get_rate_limiting(self) -> tuple[int, int]:
 # Фабрика для создания API
 class RepositoryFactory:
     @staticmethod
-    def create_api(source: str, token: str, base_url: str | None = None) -> IRepositoryAPI:
+    def create_api(
+        source: str, token: str, base_url: str | None = None
+    ) -> IRepositoryAPI:
         from GitHubRepoAPI import GitHubRepoAPI
         from ForgejoRepoAPI import ForgejoRepoAPI
 
         if source == 'github':
             return GitHubRepoAPI(Github(token))
         elif source == 'forgejo':
             if not isinstance(base_url, str):
-                raise ValueError(f"base_url for PyforgejoApi should be str, got {type(base_url)}")
+                raise ValueError(
+                    f"base_url for PyforgejoApi should be str, got {type(base_url)}"
+                )
             return ForgejoRepoAPI(PyforgejoApi(api_key=token, base_url=base_url))
         else:
             raise ValueError(f"Unsupported source: {source}")