-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathgithub_client.py
More file actions
175 lines (157 loc) · 4.93 KB
/
Copy pathgithub_client.py
File metadata and controls
175 lines (157 loc) · 4.93 KB
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
import logging
import os
import random
import shutil
from time import sleep
import pygit2
from github import Auth, Github
from github.GithubException import BadCredentialsException, GithubException
from github.Repository import Repository
from platformdirs import user_data_dir
from .file_io import _write_authors_file
from .git_client import _commit_and_push
from .markdown_builder import _create_member_file
from .strings import _
logger = logging.getLogger(__name__)
def get_repo(token: str) -> tuple[str, Repository]:
auth = Auth.Token(token)
g = Github(auth=auth)
try:
return token, g.get_repo("pythonpe/python.pe")
except BadCredentialsException as err:
raise ValueError(
_("Unauthorized access. Please check your access token.")
) from err
except GithubException as err:
raise ValueError(
_("Repository not found. Please check your access token.")
) from err
def fork_repo(token: str, original_repo: Repository) -> tuple[str, Repository]:
forked_repo = original_repo.create_fork()
forked_repo_url = forked_repo.clone_url
repo_path = user_data_dir(appname="edit-python-pe", appauthor="python.pe")
if os.path.exists(repo_path):
shutil.rmtree(repo_path, ignore_errors=True)
callbacks = pygit2.callbacks.RemoteCallbacks(
credentials=pygit2.UserPass(token, "x-oauth-basic")
)
max_retries = 5
for attempt in range(max_retries):
try:
pygit2.clone_repository(forked_repo_url, repo_path, callbacks=callbacks)
break
except Exception as e:
if attempt == max_retries - 1:
logger.error(
"Failed to clone forked repository after %d attempts", max_retries
)
raise
sleep_time = 2**attempt + random.uniform(0, 1) # noqa: S311
logger.warning(
"Attempt %d to clone repository failed: %s. Retrying in %.2fs...",
attempt + 1,
e,
sleep_time,
)
sleep(sleep_time)
return repo_path, forked_repo
def create_pr(
file_content: str,
current_file: str | None,
repo_path: str,
original_repo: Repository,
forked_repo: Repository,
token: str,
aliases: list[str],
name: str,
email: str,
) -> tuple[str, str | None]:
name_file, _unused_file_path = _create_member_file(
file_content,
current_file,
repo_path,
aliases,
name,
email,
)
_write_authors_file(
repo_path,
aliases,
name,
email,
)
# commit & push
commit_msg, repo, remote, callbacks = _commit_and_push(
repo_path,
token,
current_file is not None,
name_file,
name,
email,
)
# PR logic
pr_title = commit_msg
first_alias = aliases[0] if aliases else ""
pr_body = (
f"Changing an entry to `blog/members` for {name} (alias: {first_alias})."
if current_file
else (
f"Creating a new entry to `blog/members` for {name} (alias: {first_alias})."
)
)
fork_owner = forked_repo.owner.login
head_branch = f"{fork_owner}:main"
base_branch = "main"
pr_url = None
# If editing, retrieve PR by title and push to its branch
if current_file:
# Try to find an open PR with matching title
prs = original_repo.get_pulls(
state="open", sort="created", base=base_branch, head=head_branch
)
pr_found = None
for pr in prs:
if pr.title.endswith(current_file):
pr_found = pr
break
if pr_found:
# Push to the PR branch
remote.push([repo.head.name], callbacks=callbacks)
pr_url = pr_found.html_url
return (
_(
"Woohoo! Changes to {name_file} were successfully sent to your "
"existing PR! 🎉"
).format(name_file=name_file),
pr_url,
)
else:
pr = original_repo.create_pull(
title=pr_title,
body=pr_body,
head=head_branch,
base=base_branch,
)
pr_url = pr.html_url
return (
_(
"Woohoo! {name_file} was saved successfully and "
"your new PR is ready! 🎉"
).format(name_file=name_file),
pr_url,
)
else:
pr = original_repo.create_pull(
title=pr_title,
body=pr_body,
head=head_branch,
base=base_branch,
)
pr_url = pr.html_url
return (
_(
"Woohoo! {name_file} was saved successfully and "
"your new PR is ready! 🎉"
).format(name_file=name_file),
pr_url,
)