mirror of
https://github.com/zulip/zulip.git
synced 2025-11-02 13:03:29 +00:00
fead14951c
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>
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
# Webhooks for external integrations.
|
|
from typing import Any, Dict, List
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
|
|
from zerver.decorator import api_key_only_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
|
|
|
|
subject_types: Dict[str, List[List[str]]] = {
|
|
'app': [ # Object type name
|
|
['name'], # Title
|
|
['html_url'], # Automatically put into title
|
|
['language'], # Other properties.
|
|
['framework']
|
|
],
|
|
'base': [
|
|
['title'],
|
|
['html_url'],
|
|
['#summary'],
|
|
['subject']
|
|
],
|
|
'comment': [
|
|
[''],
|
|
['subject']
|
|
],
|
|
'errorgroup': [
|
|
['E#{}', 'number'],
|
|
['html_url'],
|
|
['last_occurrence:error']
|
|
],
|
|
'error': [
|
|
[''],
|
|
['">**Most recent Occurrence**'],
|
|
['in {}', 'extra/pathname'],
|
|
['!message']
|
|
]
|
|
}
|
|
|
|
|
|
def get_value(_obj: Dict[str, Any], key: str) -> str:
|
|
for _key in key.lstrip('!').split('/'):
|
|
if _key in _obj.keys():
|
|
_obj = _obj[_key]
|
|
else:
|
|
return ''
|
|
return str(_obj)
|
|
|
|
|
|
def format_object(
|
|
obj: Dict[str, Any],
|
|
subject_type: str,
|
|
message: str
|
|
) -> str:
|
|
if subject_type not in subject_types.keys():
|
|
return message
|
|
keys: List[List[str]] = subject_types[subject_type][1:]
|
|
title = subject_types[subject_type][0]
|
|
if title[0] != '':
|
|
title_str = ''
|
|
if len(title) > 1:
|
|
title_str = title[0].format(get_value(obj, title[1]))
|
|
else:
|
|
title_str = obj[title[0]]
|
|
if obj['html_url'] is not None:
|
|
url: str = obj['html_url']
|
|
if 'opbeat.com' not in url:
|
|
url = 'https://opbeat.com/' + url.lstrip('/')
|
|
message += '\n**[{}]({})**'.format(title_str, url)
|
|
else:
|
|
message += '\n**{}**'.format(title_str)
|
|
for key_list in keys:
|
|
if len(key_list) > 1:
|
|
value = key_list[0].format(get_value(obj, key_list[1]))
|
|
message += '\n>{}'.format(value)
|
|
else:
|
|
key = key_list[0]
|
|
key_raw = key.lstrip('!').lstrip('#').lstrip('"')
|
|
if key_raw != 'html_url' and key_raw != 'subject' and ':' not in key_raw:
|
|
value = get_value(obj, key_raw)
|
|
if key.startswith('!'):
|
|
message += '\n>{}'.format(value)
|
|
elif key.startswith('#'):
|
|
message += '\n{}'.format(value)
|
|
elif key.startswith('"'):
|
|
message += '\n{}'.format(key_raw)
|
|
else:
|
|
message += '\n>{}: {}'.format(key, value)
|
|
if key == 'subject':
|
|
message = format_object(
|
|
obj['subject'], obj['subject_type'], message + '\n')
|
|
if ':' in key:
|
|
value, value_type = key.split(':')
|
|
message = format_object(obj[value], value_type, message + '\n')
|
|
return message
|
|
|
|
|
|
@api_key_only_webhook_view("Opbeat")
|
|
@has_request_variables
|
|
def api_opbeat_webhook(request: HttpRequest, user_profile: UserProfile,
|
|
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
|
|
"""
|
|
This uses the subject name from opbeat to make the subject,
|
|
and the summary from Opbeat as the message body, with
|
|
details about the object mentioned.
|
|
"""
|
|
|
|
message_subject = payload['title']
|
|
|
|
message = format_object(payload, 'base', '')
|
|
|
|
check_send_webhook_message(request, user_profile, message_subject, message)
|
|
return json_success()
|