paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-04 14:03:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

zerver/webhooks: Use python 3 syntax for typing.

Browse Source 
Tweaked by tabbott to fix various line-wrapping issues.
This commit is contained in:
rht
2017-11-04 07:47:46 +01:00
committed by Tim Abbott
parent 68b0a419ec
commit 969cc506d2
80 changed files with 797 additions and 1508 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
zerver/webhooks/github_webhook/view.py
View File

@@ -18,8 +18,7 @@ from zerver.lib.webhooks.git import get_issue_event_message, SUBJECT_WITH_PR_OR_
class UnknownEventType(Exception):
pass
def get_opened_or_update_pull_request_body(payload):
# type: (Dict[str, Any]) -> Text
def get_opened_or_update_pull_request_body(payload: Dict[str, Any]) -> Text:
pull_request = payload['pull_request']
action = payload['action']
if action == 'synchronize':
@@ -38,8 +37,7 @@ def get_opened_or_update_pull_request_body(payload):
assignee=assignee
)
def get_assigned_or_unassigned_pull_request_body(payload):
# type: (Dict[str, Any]) -> Text
def get_assigned_or_unassigned_pull_request_body(payload: Dict[str, Any]) -> Text:
pull_request = payload['pull_request']
assignee = pull_request.get('assignee')
if assignee is not None:
@@ -54,8 +52,7 @@ def get_assigned_or_unassigned_pull_request_body(payload):
return "{} to {}".format(base_message, assignee)
return base_message
def get_closed_pull_request_body(payload):
# type: (Dict[str, Any]) -> Text
def get_closed_pull_request_body(payload: Dict[str, Any]) -> Text:
pull_request = payload['pull_request']
action = 'merged' if pull_request['merged'] else 'closed without merge'
return get_pull_request_event_message(
@@ -64,8 +61,7 @@ def get_closed_pull_request_body(payload):
pull_request['html_url'],
)
def get_membership_body(payload):
# type: (Dict[str, Any]) -> Text
def get_membership_body(payload: Dict[str, Any]) -> Text:
action = payload['action']
member = payload['member']
scope = payload['scope']
@@ -80,8 +76,7 @@ def get_membership_body(payload):
scope
)
def get_member_body(payload):
# type: (Dict[str, Any]) -> Text
def get_member_body(payload: Dict[str, Any]) -> Text:
return u"{} {} [{}]({}) to [{}]({})".format(
get_sender_name(payload),
payload['action'],
@@ -91,8 +86,7 @@ def get_member_body(payload):
payload['repository']['html_url']
)
def get_issue_body(payload):
# type: (Dict[str, Any]) -> Text
def get_issue_body(payload: Dict[str, Any]) -> Text:
action = payload['action']
issue = payload['issue']
assignee = issue['assignee']
@@ -105,8 +99,7 @@ def get_issue_body(payload):
assignee=assignee['login'] if assignee else None
)
def get_issue_comment_body(payload):
# type: (Dict[str, Any]) -> Text
def get_issue_comment_body(payload: Dict[str, Any]) -> Text:
action = payload['action']
comment = payload['comment']
issue = payload['issue']
@@ -125,8 +118,7 @@ def get_issue_comment_body(payload):
comment['body'],
)
def get_fork_body(payload):
# type: (Dict[str, Any]) -> Text
def get_fork_body(payload: Dict[str, Any]) -> Text:
forkee = payload['forkee']
return u"{} forked [{}]({})".format(
get_sender_name(payload),
@@ -134,20 +126,17 @@ def get_fork_body(payload):
forkee['html_url']
)
def get_deployment_body(payload):
# type: (Dict[str, Any]) -> Text
def get_deployment_body(payload: Dict[str, Any]) -> Text:
return u'{} created new deployment'.format(
get_sender_name(payload),
)
def get_change_deployment_status_body(payload):
# type: (Dict[str, Any]) -> Text
def get_change_deployment_status_body(payload: Dict[str, Any]) -> Text:
return u'Deployment changed status to {}'.format(
payload['deployment_status']['state'],
)
def get_create_or_delete_body(payload, action):
# type: (Dict[str, Any], Text) -> Text
def get_create_or_delete_body(payload: Dict[str, Any], action: Text) -> Text:
ref_type = payload['ref_type']
return u'{} {} {} {}'.format(
get_sender_name(payload),
@@ -156,8 +145,7 @@ def get_create_or_delete_body(payload, action):
payload['ref']
).rstrip()
def get_commit_comment_body(payload):
# type: (Dict[str, Any]) -> Text
def get_commit_comment_body(payload: Dict[str, Any]) -> Text:
comment = payload['comment']
comment_url = comment['html_url']
commit_url = comment_url.split('#', 1)[0]
@@ -170,16 +158,14 @@ def get_commit_comment_body(payload):
comment['body'],
)
def get_push_tags_body(payload):
# type: (Dict[str, Any]) -> Text
def get_push_tags_body(payload: Dict[str, Any]) -> Text:
return get_push_tag_event_message(
get_sender_name(payload),
get_tag_name_from_ref(payload['ref']),
action='pushed' if payload.get('created') else 'removed'
)
def get_push_commits_body(payload):
# type: (Dict[str, Any]) -> Text
def get_push_commits_body(payload: Dict[str, Any]) -> Text:
commits_data = [{
'name': (commit.get('author').get('username') or
commit.get('author').get('name')),
@@ -195,15 +181,13 @@ def get_push_commits_body(payload):
deleted=payload['deleted']
)
def get_public_body(payload):
# type: (Dict[str, Any]) -> Text
def get_public_body(payload: Dict[str, Any]) -> Text:
return u"{} made [the repository]({}) public".format(
get_sender_name(payload),
payload['repository']['html_url'],
)
def get_wiki_pages_body(payload):
# type: (Dict[str, Any]) -> Text
def get_wiki_pages_body(payload: Dict[str, Any]) -> Text:
wiki_page_info_template = u"* {action} [{title}]({url})\n"
wiki_info = u''
for page in payload['pages']:
@@ -214,37 +198,32 @@ def get_wiki_pages_body(payload):
)
return u"{}:\n{}".format(get_sender_name(payload), wiki_info.rstrip())
def get_watch_body(payload):
# type: (Dict[str, Any]) -> Text
def get_watch_body(payload: Dict[str, Any]) -> Text:
return u"{} starred [the repository]({})".format(
get_sender_name(payload),
payload['repository']['html_url']
)
def get_repository_body(payload):
# type: (Dict[str, Any]) -> Text
def get_repository_body(payload: Dict[str, Any]) -> Text:
return u"{} {} [the repository]({})".format(
get_sender_name(payload),
payload.get('action'),
payload['repository']['html_url']
)
def get_add_team_body(payload):
# type: (Dict[str, Any]) -> Text
def get_add_team_body(payload: Dict[str, Any]) -> Text:
return u"[The repository]({}) was added to team {}".format(
payload['repository']['html_url'],
payload['team']['name']
)
def get_release_body(payload):
# type: (Dict[str, Any]) -> Text
def get_release_body(payload: Dict[str, Any]) -> Text:
return u"{} published [the release]({})".format(
get_sender_name(payload),
payload['release']['html_url'],
)
def get_page_build_body(payload):
# type: (Dict[str, Any]) -> Text
def get_page_build_body(payload: Dict[str, Any]) -> Text:
build = payload['build']
action = build['status']
if action == 'null':
@@ -262,8 +241,7 @@ def get_page_build_body(payload):
action
)
def get_status_body(payload):
# type: (Dict[str, Any]) -> Text
def get_status_body(payload: Dict[str, Any]) -> Text:
if payload['target_url']:
status = '[{}]({})'.format(
payload['state'],
@@ -277,8 +255,7 @@ def get_status_body(payload):
status
)
def get_pull_request_review_body(payload):
# type: (Dict[str, Any]) -> Text
def get_pull_request_review_body(payload: Dict[str, Any]) -> Text:
return get_pull_request_event_message(
get_sender_name(payload),
'submitted',
@@ -286,8 +263,7 @@ def get_pull_request_review_body(payload):
type='PR Review'
)
def get_pull_request_review_comment_body(payload):
# type: (Dict[str, Any]) -> Text
def get_pull_request_review_comment_body(payload: Dict[str, Any]) -> Text:
action = payload['action']
message = None
if action == 'created':
@@ -301,36 +277,28 @@ def get_pull_request_review_comment_body(payload):
type='PR Review Comment'
)
def get_ping_body(payload):
# type: (Dict[str, Any]) -> Text
def get_ping_body(payload: Dict[str, Any]) -> Text:
return get_setup_webhook_message('GitHub', get_sender_name(payload))
def get_repository_name(payload):
# type: (Dict[str, Any]) -> Text
def get_repository_name(payload: Dict[str, Any]) -> Text:
return payload['repository']['name']
def get_organization_name(payload):
# type: (Dict[str, Any]) -> Text
def get_organization_name(payload: Dict[str, Any]) -> Text:
return payload['organization']['login']
def get_sender_name(payload):
# type: (Dict[str, Any]) -> Text
def get_sender_name(payload: Dict[str, Any]) -> Text:
return payload['sender']['login']
def get_branch_name_from_ref(ref_string):
# type: (Text) -> Text
def get_branch_name_from_ref(ref_string: Text) -> Text:
return re.sub(r'^refs/heads/', '', ref_string)
def get_tag_name_from_ref(ref_string):
# type: (Text) -> Text
def get_tag_name_from_ref(ref_string: Text) -> Text:
return re.sub(r'^refs/tags/', '', ref_string)
def is_commit_push_event(payload):
# type: (Dict[str, Any]) -> bool
def is_commit_push_event(payload: Dict[str, Any]) -> bool:
return bool(re.match(r'^refs/heads/', payload['ref']))
def get_subject_based_on_type(payload, event):
# type: (Dict[str, Any], Text) -> Text
def get_subject_based_on_type(payload: Dict[str, Any], event: Text) -> Text:
if 'pull_request' in event:
return SUBJECT_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=get_repository_name(payload),
@@ -409,8 +377,7 @@ def api_github_webhook(
check_send_stream_message(user_profile, request.client, stream, subject, body)
return json_success()
def get_event(request, payload, branches):
# type: (HttpRequest, Dict[str, Any], Text) -> Optional[str]
def get_event(request: HttpRequest, payload: Dict[str, Any], branches: Text) -> Optional[str]:
event = request.META['HTTP_X_GITHUB_EVENT']
if event == 'pull_request':
action = payload['action']
@@ -436,6 +403,5 @@ def get_event(request, payload, branches):
logging.warning(u'Event {} is unknown and cannot be handled'.format(event))
return None
def get_body_function_based_on_type(type):
# type: (str) -> Any
def get_body_function_based_on_type(type: str) -> Any:
return EVENT_FUNCTION_MAPPER.get(type)