Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
except InvalidReleaseFileError:
exit_code = 2
status = "INVALID"
mutation = """mutation AddReleaseComment($input: AddReleaseFileCommentInput!) {
addReleaseFileComment(input: $input)
}"""
mutation_input = {
"prNumber": event_data["number"],
"status": status,
"releaseInfo": release_info,
}
response = httpx.post(
API_URL,
json={"query": mutation, "variables": {"input": mutation_input}},
timeout=120,
)
response.raise_for_status()
print(f"Status is {status}")
sys.exit(exit_code)
def sim_handler(task_id, meta_param_dict, adjustment):
from cs_config import functions
outputs = functions.run_model(meta_param_dict, adjustment)
print("got result")
outputs = cs_storage.serialize_to_json(outputs)
print("storing results")
for i in range(3):
try:
resp = httpx.post(
"http://outputs-processor/write/",
json={"task_id": task_id, "outputs": outputs},
timeout=120.0,
)
break
except Exception as e:
print(i, e)
print("got resp", resp.status_code, resp.url)
assert resp.status_code == 200, f"Got code: {resp.status_code}"
return resp.json()
def trigger_appveyor_build(account: str, project: str, token: str) -> str:
"""Returns an URL for the build or raises RequestException"""
r = httpx.post(
"https://ci.appveyor.com/api/builds",
json={
"accountName": account,
"projectSlug": project,
"branch": "master",
},
headers={
"Authorization": "Bearer " + token,
},
timeout=REQUEST_TIMEOUT)
r.raise_for_status()
try:
build_id = r.json()['buildId']
except (ValueError, KeyError):
build_id = 0
def add_or_edit_comment(github_event_data: dict, comment: str):
current_comments = get_comments(github_event_data)
previous_comment = next(
(comment for comment in current_comments if is_release_check_comment(comment)),
None,
)
method = httpx.patch if previous_comment else httpx.post
url = (
previous_comment["url"]
if previous_comment
else get_comments_link(github_event_data)
)
request = method(
url,
headers={"Authorization": f"token {GITHUB_TOKEN}"},
json={"body": comment + SIGNATURE},
)
if request.status_code >= 400:
print(request.text)
print(request.status_code)
def stage_app(self, app):
resp = httpx.post(
f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
json={"staging_tag": self.staging_tag},
headers={"Authorization": f"Token {self.cs_api_token}"},
)
assert (
resp.status_code == 200
), f"Got: {resp.url} {resp.status_code} {resp.text}"
sys.stdout.write(resp.json()["staging_tag"])
current_labels_url_by_name = {
label["name"]: label["url"]
for label in github_event_data["pull_request"]["labels"]
}
current_labels = set(current_labels_url_by_name.keys())
release_labels_to_remove = [
label
for label in current_labels
if label.startswith("bot:release-type-") and label != new_release_label
]
labels_to_remove.update(release_labels_to_remove)
if not current_labels.issuperset(labels_to_add):
request = httpx.post(
labels_url,
headers={"Authorization": f"token {GITHUB_TOKEN}"},
json={"labels": list(labels_to_add)},
)
if request.status_code >= 400:
print(request.text)
print(request.status_code)
if current_labels.issuperset(labels_to_remove):
for label in labels_to_remove:
request = httpx.delete(
current_labels_url_by_name[label],
headers={"Authorization": f"token {GITHUB_TOKEN}"},
)
def promote_app(self, app):
resp = httpx.get(
f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
headers={"Authorization": f"Token {self.cs_api_token}"},
)
assert (
resp.status_code == 200
), f"Got: {resp.url} {resp.status_code} {resp.text}"
staging_tag = resp.json()["staging_tag"]
resp = httpx.post(
f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
json={"latest_tag": staging_tag or self.tag, "staging_tag": None},
headers={"Authorization": f"Token {self.cs_api_token}"},
)
assert (
resp.status_code == 200
), f"Got: {resp.url} {resp.status_code} {resp.text}"
sys.stdout.write(resp.json()["latest_tag"])