|
16 | 16 | ) |
17 | 17 |
|
18 | 18 | class ForgejoRepoAPI(IRepositoryAPI): |
19 | | - def __init__ (self, client): |
| 19 | + def __init__(self, client): |
20 | 20 | self.client = client |
21 | 21 |
|
22 | 22 | def get_user_data(self, user) -> User: |
23 | 23 | return User( |
24 | 24 | login=user.login, |
25 | | - username=getattr(user, 'full_name',"No name"), |
| 25 | + username=getattr(user, 'full_name', "No name"), |
26 | 26 | email=getattr(user, 'email', ""), |
27 | 27 | html_url=user.html_url, |
28 | 28 | node_id=user.id, |
@@ -52,19 +52,65 @@ def get_repository(self, id: str) -> Repository | None: |
52 | 52 | return None |
53 | 53 |
|
54 | 54 | def get_collaborator_permission(self, repo: Repository, user: User) -> str: |
55 | | - return " " |
| 55 | + return "|||" |
56 | 56 |
|
57 | 57 | def get_commits(self, repo: Repository, files: bool = True) -> list[Commit]: |
| 58 | + try: |
| 59 | + commits = self.client.repository.repo_get_all_commits(repo.owner.login, repo.name) |
| 60 | + return [ |
| 61 | + Commit( |
| 62 | + _id=c.sha, |
| 63 | + message=c.commit.message, |
| 64 | + author=self.get_user_data(c.author), |
| 65 | + date=c.commit.author.date, |
| 66 | + files=[f.filename for f in getattr(c, "files", [])] if files else None |
| 67 | + ) |
| 68 | + for c in commits |
| 69 | + ] |
| 70 | + except Exception as e: |
| 71 | + logging.error( |
| 72 | + f"Failed to get commits from Forgejo for repo {repo.name}: {e}" |
| 73 | + ) |
58 | 74 | return [] |
59 | 75 |
|
60 | 76 | def get_contributors(self, repo: Repository) -> list[Contributor]: |
61 | | - return [] |
| 77 | + try: |
| 78 | + commits = self.client.repository.repo_get_all_commits(repo.owner.login, repo.name) |
| 79 | + contributors = {c.author.login: c.author.email or "" for c in commits if c.author} |
| 80 | + return [Contributor(login, email) for login, email in contributors.items()] |
| 81 | + except Exception as e: |
| 82 | + logging.error(f"Failed to get contributors from Forgejo for repo {repo.name}: {e}") |
| 83 | + return [] |
62 | 84 |
|
63 | 85 | def get_issues(self, repo: Repository) -> list[Issue]: |
64 | | - return [] |
| 86 | + return [] |
65 | 87 |
|
66 | 88 | def get_pull_requests(self, repo: Repository) -> list[PullRequest]: |
67 | | - return [] |
| 89 | + try: |
| 90 | + pulls = self.client.repository.repo_list_pull_requests(repo.owner.login, repo.name) |
| 91 | + |
| 92 | + return [ |
| 93 | + PullRequest( |
| 94 | + _id=p.number, |
| 95 | + title=p.title, |
| 96 | + author=self.get_user_data(p.user), |
| 97 | + state=p.state, |
| 98 | + created_at=p.created_at, |
| 99 | + head_label=p.head.ref, |
| 100 | + base_label=p.base.ref, |
| 101 | + head_ref=p.head.ref, |
| 102 | + base_ref=p.base.ref, |
| 103 | + merged_by=self.get_user_data(p.merged_by) if p.merged_by else None, |
| 104 | + files=[file.filename for file in p.files], |
| 105 | + issue_url=p.issue_url, |
| 106 | + labels=[label.name for label in p.labels] if p.labels else [], |
| 107 | + milestone=p.milestone.title if p.milestone else None, |
| 108 | + ) |
| 109 | + for p in pulls |
| 110 | + ] |
| 111 | + except Exception as e: |
| 112 | + logging.error(f"Failed to get pull requests from Forgejo for repo {repo.name}: {e}") |
| 113 | + return [] |
68 | 114 |
|
69 | 115 | def get_branches(self, repo: Repository) -> list[Branch]: |
70 | 116 | return [] |
|
0 commit comments