mirror of
https://github.com/zulip/zulip.git
synced 2025-11-08 07:52:19 +00:00
87dcd5442a
This fixes some error message strings and skips converting request_data into json. From now, conversion would be the responsibility of interface. Also, base_url is now not passed into event structure.
33 lines
1.3 KiB
Python
33 lines
1.3 KiB
Python
from __future__ import absolute_import
|
|
from typing import Any, Dict, Text, Tuple, Callable, Mapping, Optional
|
|
import json
|
|
from requests import Response
|
|
from zerver.models import UserProfile
|
|
|
|
from zerver.lib.outgoing_webhook import OutgoingWebhookServiceInterface
|
|
|
|
class GenericOutgoingWebhookService(OutgoingWebhookServiceInterface):
|
|
|
|
def __init__(self, base_url, token, user_profile, service_name):
|
|
# type: (Text, Text, UserProfile, Text) -> None
|
|
super(GenericOutgoingWebhookService, self).__init__(base_url, token, user_profile, service_name)
|
|
|
|
def process_event(self, event):
|
|
# type: (Dict[Text, Any]) -> Tuple[Dict[str, Any], Any]
|
|
rest_operation = {'method': 'POST',
|
|
'relative_url_path': '',
|
|
'base_url': self.base_url,
|
|
'request_kwargs': {}}
|
|
request_data = {"data": event['command'],
|
|
"message": event['message'],
|
|
"token": self.token}
|
|
return rest_operation, json.dumps(request_data)
|
|
|
|
def process_success(self, response, event):
|
|
# type: (Response, Dict[Text, Any]) -> Optional[str]
|
|
return str(response.text)
|
|
|
|
def process_failure(self, response, event):
|
|
# type: (Response, Dict[Text, Any]) -> Optional[str]
|
|
return str(response.text)
|