Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def _make_request(self, method: str, url: str, url_vars: Dict[str, str],
data: Any, accept: str,
jwt: Opt[str] = None,
oauth_token: Opt[str] = None,
) -> Tuple[bytes, Opt[str]]:
"""Construct and make an HTTP request."""
if oauth_token is not None and jwt is not None:
raise ValueError("Cannot pass both oauth_token and jwt.")
filled_url = sansio.format_url(url, url_vars)
if jwt is not None:
request_headers = sansio.create_headers(
self.requester, accept=accept,
jwt=jwt)
elif oauth_token is not None:
request_headers = sansio.create_headers(
self.requester, accept=accept,
oauth_token=oauth_token)
else:
# fallback to using oauth_token
request_headers = sansio.create_headers(
self.requester, accept=accept,
oauth_token=self.oauth_token)
cached = cacheable = False
# Can't use None as a "no body" sentinel as it's a legitimate JSON type.
if data == b"":
body = b""
request_headers["content-length"] = "0"
def create_gh_pr(self, base_branch, head_branch, *, commit_message, gh_auth):
"""
Create PR in GitHub
"""
request_headers = sansio.create_headers(self.username, oauth_token=gh_auth)
title, body = normalize_commit_message(commit_message)
if not self.prefix_commit:
title = f"[{base_branch}] {title}"
data = {
"title": title,
"body": body,
"head": f"{self.username}:{head_branch}",
"base": base_branch,
"maintainer_can_modify": True,
}
url = CREATE_PR_URL_TEMPLATE.format(config=self.config)
response = requests.post(url, headers=request_headers, json=data)
if response.status_code == requests.codes.created:
click.echo(f"Backport PR created at {response.json()['html_url']}")
else:
click.echo(response.status_code)
def comment_on_pr(repo_full_name, issue_number, message):
"""
Leave a comment on a PR/Issue
"""
request_headers = sansio.create_headers(
os.environ.get("GH_USERNAME"), oauth_token=os.getenv("GH_AUTH")
)
issue_comment_url = (
f"https://api.github.com/repos/{repo_full_name}/issues/{issue_number}/comments"
)
data = {"body": message}
response = requests.post(issue_comment_url, headers=request_headers, json=data)
if response.status_code == requests.codes.created:
print(f"Commented at {response.json()['html_url']}, message: {message}")
else:
print(response.status_code)
print(response.text)
def create_gh_pr(self, base_branch, head_branch, *,
commit_message,
gh_auth):
"""
Create PR in GitHub
"""
request_headers = sansio.create_headers(
self.username, oauth_token=gh_auth)
title, body = normalize_commit_message(commit_message)
if not self.prefix_commit:
title = f"[{base_branch}] {title}"
data = {
"title": title,
"body": body,
"head": f"{self.username}:{head_branch}",
"base": base_branch,
"maintainer_can_modify": True
}
response = requests.post(CPYTHON_CREATE_PR_URL, headers=request_headers, json=data)
if response.status_code == requests.codes.created:
click.echo(f"Backport PR created at {response.json()['html_url']}")
else:
click.echo(response.status_code)
def close_issue(repo_full_name, issue_number):
username = os.environ.get("GH_USERNAME")
gh_auth = os.environ.get("GH_AUTH")
request_headers = sansio.create_headers(username, oauth_token=gh_auth)
data = {"state": "closed"}
url = f"https://api.github.com/repos/{repo_full_name}/issues/{issue_number}"
response = requests.patch(url, headers=request_headers, json=data)
if response.status_code == requests.codes.created:
print(f"PR created at {response.json()['html_url']}")
else:
print(response.status_code)
print(response.text)