zulip/zerver/webhooks/hellosign/view.py

70 lines
2.3 KiB
Python
Raw Normal View History

2017-11-16 00:43:10 +01:00
from typing import Any, Dict, List
import orjson
2017-11-16 00:43:10 +01:00
from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
IS_AWAITING_SIGNATURE = "is awaiting the signature of {awaiting_recipients}"
WAS_JUST_SIGNED_BY = "was just signed by {signed_recipients}"
BODY = "The `{contract_title}` document {actions}."
def get_message_body(payload: Dict[str, Dict[str, Any]]) -> str:
contract_title = payload["signature_request"]["title"]
python: Convert assignment type annotations to Python 3.6 style. This commit was split by tabbott; this piece covers the vast majority of files in Zulip, but excludes scripts/, tools/, and puppet/ to help ensure we at least show the right error messages for Xenial systems. We can likely further refine the remaining pieces with some testing. Generated by com2ann, with whitespace fixes and various manual fixes for runtime issues: - invoiced_through: Optional[LicenseLedger] = models.ForeignKey( + invoiced_through: Optional["LicenseLedger"] = models.ForeignKey( -_apns_client: Optional[APNsClient] = None +_apns_client: Optional["APNsClient"] = None - notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) - signup_notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) + notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) + signup_notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) - author: Optional[UserProfile] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE) + author: Optional["UserProfile"] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE) - bot_owner: Optional[UserProfile] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL) + bot_owner: Optional["UserProfile"] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL) - default_sending_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) - default_events_register_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) + default_sending_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) + default_events_register_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) -descriptors_by_handler_id: Dict[int, ClientDescriptor] = {} +descriptors_by_handler_id: Dict[int, "ClientDescriptor"] = {} -worker_classes: Dict[str, Type[QueueProcessingWorker]] = {} -queues: Dict[str, Dict[str, Type[QueueProcessingWorker]]] = {} +worker_classes: Dict[str, Type["QueueProcessingWorker"]] = {} +queues: Dict[str, Dict[str, Type["QueueProcessingWorker"]]] = {} -AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional[LDAPSearch] = None +AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional["LDAPSearch"] = None Signed-off-by: Anders Kaseorg <anders@zulipchat.com>
2020-04-22 01:09:50 +02:00
recipients: Dict[str, List[str]] = {}
signatures = payload["signature_request"]["signatures"]
for signature in signatures:
recipients.setdefault(signature["status_code"], [])
recipients[signature["status_code"]].append(signature["signer_name"])
recipients_text = ""
if recipients.get("awaiting_signature"):
recipients_text += IS_AWAITING_SIGNATURE.format(
awaiting_recipients=get_recipients_text(recipients["awaiting_signature"]),
)
if recipients.get("signed"):
text = WAS_JUST_SIGNED_BY.format(
signed_recipients=get_recipients_text(recipients["signed"]),
)
if recipients_text:
recipients_text = f"{recipients_text}, and {text}"
else:
recipients_text = text
return BODY.format(contract_title=contract_title, actions=recipients_text).strip()
def get_recipients_text(recipients: List[str]) -> str:
recipients_text = ""
if len(recipients) == 1:
recipients_text = "{}".format(*recipients)
else:
for recipient in recipients[:-1]:
recipients_text += f"{recipient}, "
recipients_text += f"and {recipients[-1]}"
return recipients_text
@webhook_view("HelloSign")
@has_request_variables
def api_hellosign_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: Dict[str, Dict[str, Any]] = REQ(whence="json", converter=orjson.loads),
) -> HttpResponse:
if "signature_request" in payload:
body = get_message_body(payload)
topic = payload["signature_request"]["title"]
check_send_webhook_message(request, user_profile, topic, body)
return json_success({"msg": "Hello API Event Received"})