Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

greenhouse: Strengthen types using WildValue.

This commit is contained in:
Hari Prashant Bhimaraju 2022-10-08 23:51:17 +05:30 committed by Tim Abbott
parent a38b1390ac
commit 08eb34d0d6
1 changed files with 22 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
zerver/webhooks/greenhouse/view.py
View File

@ -1,10 +1,16 @@
from typing import Any, Dict, List
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success from zerver.lib.response import json_success
from zerver.lib.validator import (
WildValue,
check_int,
check_none_or,
check_string,
check_url,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile from zerver.models import UserProfile
@ -16,12 +22,12 @@ MESSAGE_TEMPLATE = """
""".strip() """.strip()
def dict_list_to_string(some_list: List[Any]) -> str: def dict_list_to_string(some_list: WildValue) -> str:
internal_template = "" internal_template = ""
for item in some_list: for item in some_list:
item_type = item.get("type", "").title() item_type = item.get("type", "").tame(check_string).title()
item_value = item.get("value") item_value = item.get("value").tame(check_none_or(check_string))
item_url = item.get("url") item_url = item.get("url").tame(check_none_or(check_url))
if item_type and item_value: if item_type and item_value:
internal_template += f"{item_value} ({item_type}), " internal_template += f"{item_value} ({item_type}), "
elif item_type and item_url: elif item_type and item_url:
@ -36,29 +42,30 @@ def dict_list_to_string(some_list: List[Any]) -> str:
def api_greenhouse_webhook( def api_greenhouse_webhook(
request: HttpRequest, request: HttpRequest,
user_profile: UserProfile, user_profile: UserProfile,
payload: Dict[str, Any] = REQ(argument_type="body"), payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
) -> HttpResponse: ) -> HttpResponse:
if payload["action"] == "ping": action = payload["action"].tame(check_string)
if action == "ping":
return json_success(request) return json_success(request)
if payload["action"] == "update_candidate": if action == "update_candidate":
candidate = payload["payload"]["candidate"] candidate = payload["payload"]["candidate"]
else: else:
candidate = payload["payload"]["application"]["candidate"] candidate = payload["payload"]["application"]["candidate"]
action = payload["action"].replace("_", " ").title() action = action.replace("_", " ").title()
application = payload["payload"]["application"] application = payload["payload"]["application"]
body = MESSAGE_TEMPLATE.format( body = MESSAGE_TEMPLATE.format(
action=action, action=action,
first_name=candidate["first_name"], first_name=candidate["first_name"].tame(check_string),
last_name=candidate["last_name"], last_name=candidate["last_name"].tame(check_string),
candidate_id=str(candidate["id"]), candidate_id=str(candidate["id"].tame(check_int)),
role=application["jobs"][0]["name"], role=application["jobs"][0]["name"].tame(check_string),
emails=dict_list_to_string(application["candidate"]["email_addresses"]), emails=dict_list_to_string(application["candidate"]["email_addresses"]),
attachments=dict_list_to_string(application["candidate"]["attachments"]), attachments=dict_list_to_string(application["candidate"]["attachments"]),
) )
topic = "{} - {}".format(action, str(candidate["id"])) topic = "{} - {}".format(action, str(candidate["id"].tame(check_int)))
check_send_webhook_message(request, user_profile, topic, body) check_send_webhook_message(request, user_profile, topic, body)
return json_success(request) return json_success(request)