-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathgithub_components.py
194 lines (159 loc) · 6.66 KB
/
github_components.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
import os
from xai_components.base import InArg, OutArg, Component, xai_component
from github import Github
def get_github_client(ctx, client=None, token=None):
if client is not None:
return client
if token is not None:
return Github(token)
if ctx.get('github_client', None) is None:
ctx['github_client'] = Github(os.getenv("GITHUB_TOKEN"))
return ctx['github_client']
@xai_component
class GithubAuthorize(Component):
"""Component to authorize GitHub client using a provided token or environment variables.
##### outPorts:
- client (Github): The authenticated GitHub client.
"""
token: InArg[str]
client: OutArg[Github]
def execute(self, ctx) -> None:
self.client.value = get_github_client(ctx, client=None, token=self.token.value)
ctx['github_client'] = self.client.value
@xai_component
class GithubListIssues(Component):
"""Component to list issues in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- client (Github): The authenticated GitHub client.
##### outPorts:
- issues (list): A list of issues in the repository.
"""
repo_name: InArg[str]
client: InArg[Github]
issues: OutArg[list]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
self.issues.value = [issue.title for issue in repo.get_issues()]
@xai_component
class GithubGetIssue(Component):
"""Component to get a specific issue from a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- issue_number (int): The number of the issue to retrieve.
- client (Github): The authenticated GitHub client.
##### outPorts:
- issue (dict): The details of the issue.
"""
repo_name: InArg[str]
issue_number: InArg[int]
client: InArg[Github]
issue: OutArg[dict]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
self.issue.value = repo.get_issue(self.issue_number.value).raw_data
@xai_component
class GithubCreateIssue(Component):
"""Component to create a new issue in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- title (str): The title of the new issue.
- body (str): The body of the new issue.
- client (Github): The authenticated GitHub client.
##### outPorts:
- issue (dict): The details of the created issue.
"""
repo_name: InArg[str]
title: InArg[str]
body: InArg[str]
client: InArg[Github]
issue: OutArg[dict]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
new_issue = repo.create_issue(title=self.title.value, body=self.body.value)
self.issue.value = new_issue.raw_data
@xai_component
class GithubListPullRequests(Component):
"""Component to list pull requests in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- client (Github): The authenticated GitHub client.
##### outPorts:
- pull_requests (list): A list of pull requests in the repository.
"""
repo_name: InArg[str]
client: InArg[Github]
pull_requests: OutArg[list]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
self.pull_requests.value = [pr.title for pr in repo.get_pull_requests()]
@xai_component
class GithubCreatePullRequest(Component):
"""Component to create a new pull request in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- title (str): The title of the new pull request.
- body (str): The body of the new pull request.
- head (str): The name of the branch where your changes are implemented.
- base (str): The name of the branch you want to merge your changes into.
- client (Github): The authenticated GitHub client.
##### outPorts:
- pull_request (dict): The details of the created pull request.
"""
repo_name: InArg[str]
title: InArg[str]
body: InArg[str]
head: InArg[str]
base: InArg[str]
client: InArg[Github]
pull_request: OutArg[dict]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
new_pr = repo.create_pull(title=self.title.value, body=self.body.value, head=self.head.value, base=self.base.value)
self.pull_request.value = new_pr.raw_data
@xai_component
class GithubReadPullRequestComments(Component):
"""Component to read comments from a pull request in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- pull_request_number (int): The number of the pull request to read comments from.
- client (Github): The authenticated GitHub client.
##### outPorts:
- comments (list): A list of comments on the pull request.
"""
repo_name: InArg[str]
pull_request_number: InArg[int]
client: InArg[Github]
comments: OutArg[list]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
pr = repo.get_pull(self.pull_request_number.value)
self.comments.value = [comment.body for comment in pr.get_review_comments()]
@xai_component
class GithubAddPullRequestComment(Component):
"""Component to add a comment to a pull request in a GitHub repository.
##### inPorts:
- repo_name (str): The name of the repository in the format 'owner/repo'.
- pull_request_number (int): The number of the pull request to add a comment to.
- comment (str): The comment to add.
- client (Github): The authenticated GitHub client.
##### outPorts:
- success (bool): Indicates if the comment was added successfully.
"""
repo_name: InArg[str]
pull_request_number: InArg[int]
comment: InArg[str]
client: InArg[Github]
success: OutArg[bool]
def execute(self, ctx) -> None:
client = get_github_client(ctx, client=self.client.value)
repo = client.get_repo(self.repo_name.value)
pr = repo.get_pull(self.pull_request_number.value)
pr.create_issue_comment(self.comment.value)
self.success.value = True