| 15 | |
| 16 | |
| 17 | class WebhookClient: |
| 18 | logger = logging.getLogger(__name__) |
| 19 | |
| 20 | def __init__( |
| 21 | self, |
| 22 | url: str, |
| 23 | timeout: int = 30, |
| 24 | ssl: Optional[SSLContext] = None, |
| 25 | proxy: Optional[str] = None, |
| 26 | default_headers: Optional[Dict[str, str]] = None, |
| 27 | ): |
| 28 | self.url = url |
| 29 | self.timeout = timeout |
| 30 | self.ssl = ssl |
| 31 | self.proxy = proxy |
| 32 | self.default_headers = default_headers if default_headers else {} |
| 33 | |
| 34 | def send( |
| 35 | self, |
| 36 | *, |
| 37 | text: Optional[str] = None, |
| 38 | attachments: Optional[List[Union[Dict[str, any], Attachment]]] = None, |
| 39 | blocks: Optional[List[Union[Dict[str, any], Block]]] = None, |
| 40 | response_type: Optional[str] = None, |
| 41 | headers: Optional[Dict[str, str]] = None, |
| 42 | ) -> WebhookResponse: |
| 43 | return self.send_dict( |
| 44 | body={ |
| 45 | "text": text, |
| 46 | "attachments": attachments, |
| 47 | "blocks": blocks, |
| 48 | "response_type": response_type, |
| 49 | }, |
| 50 | headers=headers, |
| 51 | ) |
| 52 | |
| 53 | def send_dict(self, body: Dict[str, any], headers: Optional[Dict[str, str]] = None) -> WebhookResponse: |
| 54 | return self._perform_http_request( |
| 55 | body=_build_body(body), |
| 56 | headers=_build_request_headers(self.default_headers, headers), |
| 57 | ) |
| 58 | |
| 59 | def _perform_http_request(self, *, body: Dict[str, any], headers: Dict[str, str]) -> WebhookResponse: |
| 60 | body = json.dumps(body) |
| 61 | headers["Content-Type"] = "application/json;charset=utf-8" |
| 62 | |
| 63 | if self.logger.level <= logging.DEBUG: |
| 64 | self.logger.debug(f"Sending a request - url: {self.url}, body: {body}, headers: {headers}") |
| 65 | try: |
| 66 | url = self.url |
| 67 | opener: Optional[OpenerDirector] = None |
| 68 | # for security (BAN-B310) |
| 69 | if url.lower().startswith("http"): |
| 70 | req = Request(method="POST", url=url, data=body.encode("utf-8"), headers=headers) |
| 71 | if self.proxy is not None: |
| 72 | if isinstance(self.proxy, str): |
| 73 | opener = urllib.request.build_opener( |
| 74 | ProxyHandler({"http": self.proxy, "https": self.proxy}), |
no outgoing calls