github, gitlab: Replace the usage of dict.get(key) with dict[key].

This commit is contained in:
Christopher Chong 2022-04-09 09:29:33 +00:00 committed by Tim Abbott
parent 27a498e1e7
commit 94457732c1
2 changed files with 63 additions and 71 deletions

View File

@ -85,7 +85,7 @@ def get_assigned_or_unassigned_pull_request_body(helper: Helper) -> str:
pull_request = payload["pull_request"] pull_request = payload["pull_request"]
assignee = pull_request.get("assignee") assignee = pull_request.get("assignee")
if assignee: if assignee:
stringified_assignee = assignee.get("login").tame(check_string) stringified_assignee = assignee["login"].tame(check_string)
base_message = get_pull_request_event_message( base_message = get_pull_request_event_message(
get_sender_name(payload), get_sender_name(payload),
@ -224,7 +224,7 @@ def get_commit_comment_body(helper: Helper) -> str:
get_sender_name(payload), get_sender_name(payload),
action, action,
commit_url, commit_url,
comment.get("commit_id").tame(check_string), comment["commit_id"].tame(check_string),
comment["body"].tame(check_string), comment["body"].tame(check_string),
) )
@ -234,7 +234,7 @@ def get_push_tags_body(helper: Helper) -> str:
return get_push_tag_event_message( return get_push_tag_event_message(
get_sender_name(payload), get_sender_name(payload),
get_tag_name_from_ref(payload["ref"].tame(check_string)), get_tag_name_from_ref(payload["ref"].tame(check_string)),
action="pushed" if payload.get("created").tame(check_bool) else "removed", action="pushed" if payload["created"].tame(check_bool) else "removed",
) )
@ -242,10 +242,10 @@ def get_push_commits_body(helper: Helper) -> str:
payload = helper.payload payload = helper.payload
commits_data = [] commits_data = []
for commit in payload["commits"]: for commit in payload["commits"]:
if commit.get("author").get("username"): if commit["author"].get("username"):
name = commit.get("author").get("username").tame(check_string) name = commit["author"]["username"].tame(check_string)
else: else:
name = commit.get("author").get("name").tame(check_string) name = commit["author"]["name"].tame(check_string)
commits_data.append( commits_data.append(
{ {
"name": name, "name": name,
@ -321,7 +321,7 @@ def get_repository_body(helper: Helper) -> str:
payload = helper.payload payload = helper.payload
return "{} {} the repository [{}]({}).".format( return "{} {} the repository [{}]({}).".format(
get_sender_name(payload), get_sender_name(payload),
payload.get("action").tame(check_string), payload["action"].tame(check_string),
get_repository_full_name(payload), get_repository_full_name(payload),
payload["repository"]["html_url"].tame(check_string), payload["repository"]["html_url"].tame(check_string),
) )

View File

@ -54,10 +54,10 @@ def get_normal_push_event_body(payload: WildValue) -> str:
commits = [ commits = [
{ {
"name": commit.get("author").get("name").tame(check_string), "name": commit["author"]["name"].tame(check_string),
"sha": commit.get("id").tame(check_string), "sha": commit["id"].tame(check_string),
"message": commit.get("message").tame(check_string), "message": commit["message"].tame(check_string),
"url": commit.get("url").tame(check_string), "url": commit["url"].tame(check_string),
} }
for commit in payload["commits"] for commit in payload["commits"]
] ]
@ -99,12 +99,10 @@ def get_issue_created_event_body(payload: WildValue, include_title: bool = False
get_issue_user_name(payload), get_issue_user_name(payload),
"created", "created",
get_object_url(payload), get_object_url(payload),
payload["object_attributes"].get("iid").tame(check_int), payload["object_attributes"]["iid"].tame(check_int),
stringified_description, stringified_description,
assignees=replace_assignees_username_with_name(get_assignees(payload)), assignees=replace_assignees_username_with_name(get_assignees(payload)),
title=payload["object_attributes"].get("title").tame(check_string) title=payload["object_attributes"]["title"].tame(check_string) if include_title else None,
if include_title
else None,
) )
@ -113,10 +111,8 @@ def get_issue_event_body(payload: WildValue, action: str, include_title: bool =
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
get_object_url(payload), get_object_url(payload),
payload["object_attributes"].get("iid").tame(check_int), payload["object_attributes"]["iid"].tame(check_int),
title=payload["object_attributes"].get("title").tame(check_string) title=payload["object_attributes"]["title"].tame(check_string) if include_title else None,
if include_title
else None,
) )
@ -142,12 +138,10 @@ def get_merge_request_event_body(
return get_pull_request_event_message( return get_pull_request_event_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
pull_request.get("url").tame(check_string), pull_request["url"].tame(check_string),
pull_request.get("iid").tame(check_int), pull_request["iid"].tame(check_int),
type="MR", type="MR",
title=payload["object_attributes"].get("title").tame(check_string) title=payload["object_attributes"]["title"].tame(check_string) if include_title else None,
if include_title
else None,
) )
@ -158,16 +152,14 @@ def get_merge_request_open_or_updated_body(
return get_pull_request_event_message( return get_pull_request_event_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
pull_request.get("url").tame(check_string), pull_request["url"].tame(check_string),
pull_request.get("iid").tame(check_int), pull_request["iid"].tame(check_int),
pull_request.get("source_branch").tame(check_string), pull_request["source_branch"].tame(check_string),
pull_request.get("target_branch").tame(check_string), pull_request["target_branch"].tame(check_string),
pull_request.get("description").tame(check_string), pull_request["description"].tame(check_string),
assignees=replace_assignees_username_with_name(get_assignees(payload)), assignees=replace_assignees_username_with_name(get_assignees(payload)),
type="MR", type="MR",
title=payload["object_attributes"].get("title").tame(check_string) title=payload["object_attributes"]["title"].tame(check_string) if include_title else None,
if include_title
else None,
) )
@ -206,8 +198,8 @@ def get_commented_commit_event_body(payload: WildValue) -> str:
return get_commits_comment_action_message( return get_commits_comment_action_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
payload["commit"].get("url").tame(check_string), payload["commit"]["url"].tame(check_string),
payload["commit"].get("id").tame(check_string), payload["commit"]["id"].tame(check_string),
comment["note"].tame(check_string), comment["note"].tame(check_string),
) )
@ -216,18 +208,18 @@ def get_commented_merge_request_event_body(payload: WildValue, include_title: bo
comment = payload["object_attributes"] comment = payload["object_attributes"]
action = "[commented]({}) on".format(comment["url"].tame(check_string)) action = "[commented]({}) on".format(comment["url"].tame(check_string))
url = "{}/merge_requests/{}".format( url = "{}/merge_requests/{}".format(
payload["project"].get("web_url").tame(check_string), payload["project"]["web_url"].tame(check_string),
payload["merge_request"].get("iid").tame(check_int), payload["merge_request"]["iid"].tame(check_int),
) )
return get_pull_request_event_message( return get_pull_request_event_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
url, url,
payload["merge_request"].get("iid").tame(check_int), payload["merge_request"]["iid"].tame(check_int),
message=comment["note"].tame(check_string), message=comment["note"].tame(check_string),
type="MR", type="MR",
title=payload["merge_request"].get("title").tame(check_string) if include_title else None, title=payload["merge_request"]["title"].tame(check_string) if include_title else None,
) )
@ -235,18 +227,18 @@ def get_commented_issue_event_body(payload: WildValue, include_title: bool = Fal
comment = payload["object_attributes"] comment = payload["object_attributes"]
action = "[commented]({}) on".format(comment["url"].tame(check_string)) action = "[commented]({}) on".format(comment["url"].tame(check_string))
url = "{}/issues/{}".format( url = "{}/issues/{}".format(
payload["project"].get("web_url").tame(check_string), payload["project"]["web_url"].tame(check_string),
payload["issue"].get("iid").tame(check_int), payload["issue"]["iid"].tame(check_int),
) )
return get_pull_request_event_message( return get_pull_request_event_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
url, url,
payload["issue"].get("iid").tame(check_int), payload["issue"]["iid"].tame(check_int),
message=comment["note"].tame(check_string), message=comment["note"].tame(check_string),
type="issue", type="issue",
title=payload["issue"].get("title").tame(check_string) if include_title else None, title=payload["issue"]["title"].tame(check_string) if include_title else None,
) )
@ -254,18 +246,18 @@ def get_commented_snippet_event_body(payload: WildValue, include_title: bool = F
comment = payload["object_attributes"] comment = payload["object_attributes"]
action = "[commented]({}) on".format(comment["url"].tame(check_string)) action = "[commented]({}) on".format(comment["url"].tame(check_string))
url = "{}/snippets/{}".format( url = "{}/snippets/{}".format(
payload["project"].get("web_url").tame(check_string), payload["project"]["web_url"].tame(check_string),
payload["snippet"].get("id").tame(check_int), payload["snippet"]["id"].tame(check_int),
) )
return get_pull_request_event_message( return get_pull_request_event_message(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
url, url,
payload["snippet"].get("id").tame(check_int), payload["snippet"]["id"].tame(check_int),
message=comment["note"].tame(check_string), message=comment["note"].tame(check_string),
type="snippet", type="snippet",
title=payload["snippet"].get("title").tame(check_string) if include_title else None, title=payload["snippet"]["title"].tame(check_string) if include_title else None,
) )
@ -273,13 +265,13 @@ def get_wiki_page_event_body(payload: WildValue, action: str) -> str:
return '{} {} [wiki page "{}"]({}).'.format( return '{} {} [wiki page "{}"]({}).'.format(
get_issue_user_name(payload), get_issue_user_name(payload),
action, action,
payload["object_attributes"].get("title").tame(check_string), payload["object_attributes"]["title"].tame(check_string),
payload["object_attributes"].get("url").tame(check_string), payload["object_attributes"]["url"].tame(check_string),
) )
def get_build_hook_event_body(payload: WildValue) -> str: def get_build_hook_event_body(payload: WildValue) -> str:
build_status = payload.get("build_status").tame(check_string) build_status = payload["build_status"].tame(check_string)
if build_status == "created": if build_status == "created":
action = "was created" action = "was created"
elif build_status == "running": elif build_status == "running":
@ -287,8 +279,8 @@ def get_build_hook_event_body(payload: WildValue) -> str:
else: else:
action = f"changed status to {build_status}" action = f"changed status to {build_status}"
return "Build {} from {} stage {}.".format( return "Build {} from {} stage {}.".format(
payload.get("build_name").tame(check_string), payload["build_name"].tame(check_string),
payload.get("build_stage").tame(check_string), payload["build_stage"].tame(check_string),
action, action,
) )
@ -298,7 +290,7 @@ def get_test_event_body(payload: WildValue) -> str:
def get_pipeline_event_body(payload: WildValue) -> str: def get_pipeline_event_body(payload: WildValue) -> str:
pipeline_status = payload["object_attributes"].get("status").tame(check_string) pipeline_status = payload["object_attributes"]["status"].tame(check_string)
if pipeline_status == "pending": if pipeline_status == "pending":
action = "was created" action = "was created"
elif pipeline_status == "running": elif pipeline_status == "running":
@ -309,14 +301,14 @@ def get_pipeline_event_body(payload: WildValue) -> str:
project_homepage = get_project_homepage(payload) project_homepage = get_project_homepage(payload)
pipeline_url = "{}/pipelines/{}".format( pipeline_url = "{}/pipelines/{}".format(
project_homepage, project_homepage,
payload["object_attributes"].get("id").tame(check_int), payload["object_attributes"]["id"].tame(check_int),
) )
builds_status = "" builds_status = ""
for build in payload["builds"]: for build in payload["builds"]:
build_url = "{}/-/jobs/{}".format( build_url = "{}/-/jobs/{}".format(
project_homepage, project_homepage,
build.get("id").tame(check_int), build["id"].tame(check_int),
) )
artifact_filename = build.get("artifacts_file", {}).get("filename", None) artifact_filename = build.get("artifacts_file", {}).get("filename", None)
if artifact_filename: if artifact_filename:
@ -326,13 +318,13 @@ def get_pipeline_event_body(payload: WildValue) -> str:
else: else:
artifact_string = "" artifact_string = ""
builds_status += "* [{}]({}) - {}\n{}".format( builds_status += "* [{}]({}) - {}\n{}".format(
build.get("name").tame(check_string), build["name"].tame(check_string),
build_url, build_url,
build.get("status").tame(check_string), build["status"].tame(check_string),
artifact_string, artifact_string,
) )
return "[Pipeline ({})]({}) {} with build(s):\n{}.".format( return "[Pipeline ({})]({}) {} with build(s):\n{}.".format(
payload["object_attributes"].get("id").tame(check_int), payload["object_attributes"]["id"].tame(check_int),
pipeline_url, pipeline_url,
action, action,
builds_status[:-1], builds_status[:-1],
@ -451,19 +443,19 @@ def get_subject_based_on_event(
return f"{get_repo_name(payload)} / {get_branch_name(payload)}" return f"{get_repo_name(payload)} / {get_branch_name(payload)}"
elif event == "Job Hook" or event == "Build Hook": elif event == "Job Hook" or event == "Build Hook":
return "{} / {}".format( return "{} / {}".format(
payload["repository"].get("name").tame(check_string), get_branch_name(payload) payload["repository"]["name"].tame(check_string), get_branch_name(payload)
) )
elif event == "Pipeline Hook": elif event == "Pipeline Hook":
return "{} / {}".format( return "{} / {}".format(
get_repo_name(payload), get_repo_name(payload),
payload["object_attributes"].get("ref").tame(check_string).replace("refs/heads/", ""), payload["object_attributes"]["ref"].tame(check_string).replace("refs/heads/", ""),
) )
elif event.startswith("Merge Request Hook"): elif event.startswith("Merge Request Hook"):
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format( return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repo_name(payload), repo=get_repo_name(payload),
type="MR", type="MR",
id=payload["object_attributes"].get("iid").tame(check_int), id=payload["object_attributes"]["iid"].tame(check_int),
title=payload["object_attributes"].get("title").tame(check_string) title=payload["object_attributes"]["title"].tame(check_string)
if use_merge_request_title if use_merge_request_title
else "", else "",
) )
@ -471,22 +463,22 @@ def get_subject_based_on_event(
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format( return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repo_name(payload), repo=get_repo_name(payload),
type="issue", type="issue",
id=payload["object_attributes"].get("iid").tame(check_int), id=payload["object_attributes"]["iid"].tame(check_int),
title=payload["object_attributes"].get("title").tame(check_string), title=payload["object_attributes"]["title"].tame(check_string),
) )
elif event == "Note Hook Issue" or event == "Confidential Note Hook Issue": elif event == "Note Hook Issue" or event == "Confidential Note Hook Issue":
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format( return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repo_name(payload), repo=get_repo_name(payload),
type="issue", type="issue",
id=payload["issue"].get("iid").tame(check_int), id=payload["issue"]["iid"].tame(check_int),
title=payload["issue"].get("title").tame(check_string), title=payload["issue"]["title"].tame(check_string),
) )
elif event == "Note Hook MergeRequest": elif event == "Note Hook MergeRequest":
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format( return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repo_name(payload), repo=get_repo_name(payload),
type="MR", type="MR",
id=payload["merge_request"].get("iid").tame(check_int), id=payload["merge_request"]["iid"].tame(check_int),
title=payload["merge_request"].get("title").tame(check_string) title=payload["merge_request"]["title"].tame(check_string)
if use_merge_request_title if use_merge_request_title
else "", else "",
) )
@ -495,8 +487,8 @@ def get_subject_based_on_event(
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format( return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repo_name(payload), repo=get_repo_name(payload),
type="snippet", type="snippet",
id=payload["snippet"].get("id").tame(check_int), id=payload["snippet"]["id"].tame(check_int),
title=payload["snippet"].get("title").tame(check_string), title=payload["snippet"]["title"].tame(check_string),
) )
return get_repo_name(payload) return get_repo_name(payload)
@ -505,14 +497,14 @@ def get_event(request: HttpRequest, payload: WildValue, branches: Optional[str])
event = validate_extract_webhook_http_header(request, "X_GITLAB_EVENT", "GitLab") event = validate_extract_webhook_http_header(request, "X_GITLAB_EVENT", "GitLab")
if event == "System Hook": if event == "System Hook":
# Convert the event name to a GitLab event title # Convert the event name to a GitLab event title
event_name = payload.get("event_name", payload.get("object_kind")).tame(check_string) event_name = payload.get("event_name", payload["object_kind"]).tame(check_string)
event = event_name.split("__")[0].replace("_", " ").title() event = event_name.split("__")[0].replace("_", " ").title()
event = f"{event} Hook" event = f"{event} Hook"
if event in ["Confidential Issue Hook", "Issue Hook", "Merge Request Hook", "Wiki Page Hook"]: if event in ["Confidential Issue Hook", "Issue Hook", "Merge Request Hook", "Wiki Page Hook"]:
action = payload["object_attributes"].get("action", "open").tame(check_string) action = payload["object_attributes"].get("action", "open").tame(check_string)
event = f"{event} {action}" event = f"{event} {action}"
elif event in ["Confidential Note Hook", "Note Hook"]: elif event in ["Confidential Note Hook", "Note Hook"]:
action = payload["object_attributes"].get("noteable_type").tame(check_string) action = payload["object_attributes"]["noteable_type"].tame(check_string)
event = f"{event} {action}" event = f"{event} {action}"
elif event == "Push Hook": elif event == "Push Hook":
if branches is not None: if branches is not None: