Module slack_sdk.webhook.client
Expand source code
import json
import logging
import urllib
from http.client import HTTPResponse
from ssl import SSLContext
from typing import Dict, Union, Sequence, Optional, List, Any
from urllib.error import HTTPError
from urllib.request import Request, urlopen, OpenerDirector, ProxyHandler, HTTPSHandler
from slack_sdk.errors import SlackRequestError
from slack_sdk.models.attachments import Attachment
from slack_sdk.models.blocks import Block
from .internal_utils import (
_build_body,
_build_request_headers,
_debug_log_response,
get_user_agent,
)
from .webhook_response import WebhookResponse
from slack_sdk.http_retry import default_retry_handlers
from slack_sdk.http_retry.handler import RetryHandler
from slack_sdk.http_retry.request import HttpRequest as RetryHttpRequest
from slack_sdk.http_retry.response import HttpResponse as RetryHttpResponse
from slack_sdk.http_retry.state import RetryState
from ..proxy_env_variable_loader import load_http_proxy_from_env
class WebhookClient:
url: str
timeout: int
ssl: Optional[SSLContext]
proxy: Optional[str]
default_headers: Dict[str, str]
logger: logging.Logger
retry_handlers: List[RetryHandler]
def __init__(
self,
url: str,
timeout: int = 30,
ssl: Optional[SSLContext] = None,
proxy: Optional[str] = None,
default_headers: Optional[Dict[str, str]] = None,
user_agent_prefix: Optional[str] = None,
user_agent_suffix: Optional[str] = None,
logger: Optional[logging.Logger] = None,
retry_handlers: Optional[List[RetryHandler]] = None,
):
"""API client for Incoming Webhooks and `response_url`
https://api.slack.com/messaging/webhooks
Args:
url: Complete URL to send data (e.g., `https://hooks.slack.com/XXX`)
timeout: Request timeout (in seconds)
ssl: `ssl.SSLContext` to use for requests
proxy: Proxy URL (e.g., `localhost:9000`, `http://localhost:9000`)
default_headers: Request headers to add to all requests
user_agent_prefix: Prefix for User-Agent header value
user_agent_suffix: Suffix for User-Agent header value
logger: Custom logger
retry_handlers: Retry handlers
"""
self.url = url
self.timeout = timeout
self.ssl = ssl
self.proxy = proxy
self.default_headers = default_headers if default_headers else {}
self.default_headers["User-Agent"] = get_user_agent(user_agent_prefix, user_agent_suffix)
self.logger = logger if logger is not None else logging.getLogger(__name__)
self.retry_handlers = retry_handlers if retry_handlers is not None else default_retry_handlers()
if self.proxy is None or len(self.proxy.strip()) == 0:
env_variable = load_http_proxy_from_env(self.logger)
if env_variable is not None:
self.proxy = env_variable
def send(
self,
*,
text: Optional[str] = None,
attachments: Optional[Sequence[Union[Dict[str, Any], Attachment]]] = None,
blocks: Optional[Sequence[Union[Dict[str, Any], Block]]] = None,
response_type: Optional[str] = None,
replace_original: Optional[bool] = None,
delete_original: Optional[bool] = None,
unfurl_links: Optional[bool] = None,
unfurl_media: Optional[bool] = None,
metadata: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> WebhookResponse:
"""Performs a Slack API request and returns the result.
Args:
text: The text message
(even when having blocks, setting this as well is recommended as it works as fallback)
attachments: A collection of attachments
blocks: A collection of Block Kit UI components
response_type: The type of message (either 'in_channel' or 'ephemeral')
replace_original: True if you use this option for response_url requests
delete_original: True if you use this option for response_url requests
unfurl_links: Option to indicate whether text url should unfurl
unfurl_media: Option to indicate whether media url should unfurl
metadata: Metadata attached to the message
headers: Request headers to append only for this request
Returns:
Webhook response
"""
return self.send_dict(
# It's fine to have None value elements here
# because _build_body() filters them out when constructing the actual body data
body={
"text": text,
"attachments": attachments,
"blocks": blocks,
"response_type": response_type,
"replace_original": replace_original,
"delete_original": delete_original,
"unfurl_links": unfurl_links,
"unfurl_media": unfurl_media,
"metadata": metadata,
},
headers=headers,
)
def send_dict(self, body: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> WebhookResponse:
"""Performs a Slack API request and returns the result.
Args:
body: JSON data structure (it's still a dict at this point),
if you give this argument, body_params and files will be skipped
headers: Request headers to append only for this request
Returns:
Webhook response
"""
return self._perform_http_request(
body=_build_body(body),
headers=_build_request_headers(self.default_headers, headers),
)
def _perform_http_request(self, *, body: Dict[str, Any], headers: Dict[str, str]) -> WebhookResponse:
body = json.dumps(body)
headers["Content-Type"] = "application/json;charset=utf-8"
if self.logger.level <= logging.DEBUG:
self.logger.debug(f"Sending a request - url: {self.url}, body: {body}, headers: {headers}")
url = self.url
# NOTE: Intentionally ignore the `http_verb` here
# Slack APIs accepts any API method requests with POST methods
req = Request(method="POST", url=url, data=body.encode("utf-8"), headers=headers)
resp = None
last_error = None
retry_state = RetryState()
counter_for_safety = 0
while counter_for_safety < 100:
counter_for_safety += 1
# If this is a retry, the next try started here. We can reset the flag.
retry_state.next_attempt_requested = False
try:
resp = self._perform_http_request_internal(url, req)
# The resp is a 200 OK response
return resp
except HTTPError as e:
# read the response body here
charset = e.headers.get_content_charset() or "utf-8"
response_body: str = e.read().decode(charset)
# As adding new values to HTTPError#headers can be ignored, building a new dict object here
response_headers = dict(e.headers.items())
resp = WebhookResponse(
url=url,
status_code=e.code,
body=response_body,
headers=response_headers,
)
if e.code == 429:
# for backward-compatibility with WebClient (v.2.5.0 or older)
if "retry-after" not in resp.headers and "Retry-After" in resp.headers:
resp.headers["retry-after"] = resp.headers["Retry-After"]
if "Retry-After" not in resp.headers and "retry-after" in resp.headers:
resp.headers["Retry-After"] = resp.headers["retry-after"]
_debug_log_response(self.logger, resp)
# Try to find a retry handler for this error
retry_request = RetryHttpRequest.from_urllib_http_request(req)
retry_response = RetryHttpResponse(
status_code=e.code,
headers={k: [v] for k, v in e.headers.items()},
data=response_body.encode("utf-8") if response_body is not None else None,
)
for handler in self.retry_handlers:
if handler.can_retry(
state=retry_state,
request=retry_request,
response=retry_response,
error=e,
):
if self.logger.level <= logging.DEBUG:
self.logger.info(
f"A retry handler found: {type(handler).__name__} for {req.method} {req.full_url} - {e}"
)
handler.prepare_for_next_attempt(
state=retry_state,
request=retry_request,
response=retry_response,
error=e,
)
break
if retry_state.next_attempt_requested is False:
return resp
except Exception as err:
last_error = err
self.logger.error(f"Failed to send a request to Slack API server: {err}")
# Try to find a retry handler for this error
retry_request = RetryHttpRequest.from_urllib_http_request(req)
for handler in self.retry_handlers:
if handler.can_retry(
state=retry_state,
request=retry_request,
response=None,
error=err,
):
if self.logger.level <= logging.DEBUG:
self.logger.info(
f"A retry handler found: {type(handler).__name__} for {req.method} {req.full_url} - {err}"
)
handler.prepare_for_next_attempt(
state=retry_state,
request=retry_request,
response=None,
error=err,
)
self.logger.info(f"Going to retry the same request: {req.method} {req.full_url}")
break
if retry_state.next_attempt_requested is False:
raise err
if resp is not None:
return resp
raise last_error
def _perform_http_request_internal(self, url: str, req: Request):
opener: Optional[OpenerDirector] = None
# for security (BAN-B310)
if url.lower().startswith("http"):
if self.proxy is not None:
if isinstance(self.proxy, str):
opener = urllib.request.build_opener(
ProxyHandler({"http": self.proxy, "https": self.proxy}),
HTTPSHandler(context=self.ssl),
)
else:
raise SlackRequestError(f"Invalid proxy detected: {self.proxy} must be a str value")
else:
raise SlackRequestError(f"Invalid URL detected: {url}")
# NOTE: BAN-B310 is already checked above
http_resp: Optional[HTTPResponse] = None
if opener:
http_resp = opener.open(req, timeout=self.timeout) # skipcq: BAN-B310
else:
http_resp = urlopen(req, context=self.ssl, timeout=self.timeout) # skipcq: BAN-B310
charset: str = http_resp.headers.get_content_charset() or "utf-8"
response_body: str = http_resp.read().decode(charset)
resp = WebhookResponse(
url=url,
status_code=http_resp.status,
body=response_body,
headers=http_resp.headers,
)
_debug_log_response(self.logger, resp)
return resp
Classes
class WebhookClient (url: str, timeout: int = 30, ssl: Optional[ssl.SSLContext] = None, proxy: Optional[str] = None, default_headers: Optional[Dict[str, str]] = None, user_agent_prefix: Optional[str] = None, user_agent_suffix: Optional[str] = None, logger: Optional[logging.Logger] = None, retry_handlers: Optional[List[RetryHandler]] = None)
-
API client for Incoming Webhooks and
response_url
https://api.slack.com/messaging/webhooks
Args
url
- Complete URL to send data (e.g.,
https://hooks.slack.com/XXX
) timeout
- Request timeout (in seconds)
ssl
ssl.SSLContext
to use for requestsproxy
- Proxy URL (e.g.,
localhost:9000
,http://localhost:9000
) default_headers
- Request headers to add to all requests
user_agent_prefix
- Prefix for User-Agent header value
user_agent_suffix
- Suffix for User-Agent header value
logger
- Custom logger
retry_handlers
- Retry handlers
Expand source code
class WebhookClient: url: str timeout: int ssl: Optional[SSLContext] proxy: Optional[str] default_headers: Dict[str, str] logger: logging.Logger retry_handlers: List[RetryHandler] def __init__( self, url: str, timeout: int = 30, ssl: Optional[SSLContext] = None, proxy: Optional[str] = None, default_headers: Optional[Dict[str, str]] = None, user_agent_prefix: Optional[str] = None, user_agent_suffix: Optional[str] = None, logger: Optional[logging.Logger] = None, retry_handlers: Optional[List[RetryHandler]] = None, ): """API client for Incoming Webhooks and `response_url` https://api.slack.com/messaging/webhooks Args: url: Complete URL to send data (e.g., `https://hooks.slack.com/XXX`) timeout: Request timeout (in seconds) ssl: `ssl.SSLContext` to use for requests proxy: Proxy URL (e.g., `localhost:9000`, `http://localhost:9000`) default_headers: Request headers to add to all requests user_agent_prefix: Prefix for User-Agent header value user_agent_suffix: Suffix for User-Agent header value logger: Custom logger retry_handlers: Retry handlers """ self.url = url self.timeout = timeout self.ssl = ssl self.proxy = proxy self.default_headers = default_headers if default_headers else {} self.default_headers["User-Agent"] = get_user_agent(user_agent_prefix, user_agent_suffix) self.logger = logger if logger is not None else logging.getLogger(__name__) self.retry_handlers = retry_handlers if retry_handlers is not None else default_retry_handlers() if self.proxy is None or len(self.proxy.strip()) == 0: env_variable = load_http_proxy_from_env(self.logger) if env_variable is not None: self.proxy = env_variable def send( self, *, text: Optional[str] = None, attachments: Optional[Sequence[Union[Dict[str, Any], Attachment]]] = None, blocks: Optional[Sequence[Union[Dict[str, Any], Block]]] = None, response_type: Optional[str] = None, replace_original: Optional[bool] = None, delete_original: Optional[bool] = None, unfurl_links: Optional[bool] = None, unfurl_media: Optional[bool] = None, metadata: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, ) -> WebhookResponse: """Performs a Slack API request and returns the result. Args: text: The text message (even when having blocks, setting this as well is recommended as it works as fallback) attachments: A collection of attachments blocks: A collection of Block Kit UI components response_type: The type of message (either 'in_channel' or 'ephemeral') replace_original: True if you use this option for response_url requests delete_original: True if you use this option for response_url requests unfurl_links: Option to indicate whether text url should unfurl unfurl_media: Option to indicate whether media url should unfurl metadata: Metadata attached to the message headers: Request headers to append only for this request Returns: Webhook response """ return self.send_dict( # It's fine to have None value elements here # because _build_body() filters them out when constructing the actual body data body={ "text": text, "attachments": attachments, "blocks": blocks, "response_type": response_type, "replace_original": replace_original, "delete_original": delete_original, "unfurl_links": unfurl_links, "unfurl_media": unfurl_media, "metadata": metadata, }, headers=headers, ) def send_dict(self, body: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> WebhookResponse: """Performs a Slack API request and returns the result. Args: body: JSON data structure (it's still a dict at this point), if you give this argument, body_params and files will be skipped headers: Request headers to append only for this request Returns: Webhook response """ return self._perform_http_request( body=_build_body(body), headers=_build_request_headers(self.default_headers, headers), ) def _perform_http_request(self, *, body: Dict[str, Any], headers: Dict[str, str]) -> WebhookResponse: body = json.dumps(body) headers["Content-Type"] = "application/json;charset=utf-8" if self.logger.level <= logging.DEBUG: self.logger.debug(f"Sending a request - url: {self.url}, body: {body}, headers: {headers}") url = self.url # NOTE: Intentionally ignore the `http_verb` here # Slack APIs accepts any API method requests with POST methods req = Request(method="POST", url=url, data=body.encode("utf-8"), headers=headers) resp = None last_error = None retry_state = RetryState() counter_for_safety = 0 while counter_for_safety < 100: counter_for_safety += 1 # If this is a retry, the next try started here. We can reset the flag. retry_state.next_attempt_requested = False try: resp = self._perform_http_request_internal(url, req) # The resp is a 200 OK response return resp except HTTPError as e: # read the response body here charset = e.headers.get_content_charset() or "utf-8" response_body: str = e.read().decode(charset) # As adding new values to HTTPError#headers can be ignored, building a new dict object here response_headers = dict(e.headers.items()) resp = WebhookResponse( url=url, status_code=e.code, body=response_body, headers=response_headers, ) if e.code == 429: # for backward-compatibility with WebClient (v.2.5.0 or older) if "retry-after" not in resp.headers and "Retry-After" in resp.headers: resp.headers["retry-after"] = resp.headers["Retry-After"] if "Retry-After" not in resp.headers and "retry-after" in resp.headers: resp.headers["Retry-After"] = resp.headers["retry-after"] _debug_log_response(self.logger, resp) # Try to find a retry handler for this error retry_request = RetryHttpRequest.from_urllib_http_request(req) retry_response = RetryHttpResponse( status_code=e.code, headers={k: [v] for k, v in e.headers.items()}, data=response_body.encode("utf-8") if response_body is not None else None, ) for handler in self.retry_handlers: if handler.can_retry( state=retry_state, request=retry_request, response=retry_response, error=e, ): if self.logger.level <= logging.DEBUG: self.logger.info( f"A retry handler found: {type(handler).__name__} for {req.method} {req.full_url} - {e}" ) handler.prepare_for_next_attempt( state=retry_state, request=retry_request, response=retry_response, error=e, ) break if retry_state.next_attempt_requested is False: return resp except Exception as err: last_error = err self.logger.error(f"Failed to send a request to Slack API server: {err}") # Try to find a retry handler for this error retry_request = RetryHttpRequest.from_urllib_http_request(req) for handler in self.retry_handlers: if handler.can_retry( state=retry_state, request=retry_request, response=None, error=err, ): if self.logger.level <= logging.DEBUG: self.logger.info( f"A retry handler found: {type(handler).__name__} for {req.method} {req.full_url} - {err}" ) handler.prepare_for_next_attempt( state=retry_state, request=retry_request, response=None, error=err, ) self.logger.info(f"Going to retry the same request: {req.method} {req.full_url}") break if retry_state.next_attempt_requested is False: raise err if resp is not None: return resp raise last_error def _perform_http_request_internal(self, url: str, req: Request): opener: Optional[OpenerDirector] = None # for security (BAN-B310) if url.lower().startswith("http"): if self.proxy is not None: if isinstance(self.proxy, str): opener = urllib.request.build_opener( ProxyHandler({"http": self.proxy, "https": self.proxy}), HTTPSHandler(context=self.ssl), ) else: raise SlackRequestError(f"Invalid proxy detected: {self.proxy} must be a str value") else: raise SlackRequestError(f"Invalid URL detected: {url}") # NOTE: BAN-B310 is already checked above http_resp: Optional[HTTPResponse] = None if opener: http_resp = opener.open(req, timeout=self.timeout) # skipcq: BAN-B310 else: http_resp = urlopen(req, context=self.ssl, timeout=self.timeout) # skipcq: BAN-B310 charset: str = http_resp.headers.get_content_charset() or "utf-8" response_body: str = http_resp.read().decode(charset) resp = WebhookResponse( url=url, status_code=http_resp.status, body=response_body, headers=http_resp.headers, ) _debug_log_response(self.logger, resp) return resp
Class variables
var default_headers : Dict[str, str]
var logger : logging.Logger
var proxy : Optional[str]
var retry_handlers : List[RetryHandler]
var ssl : Optional[ssl.SSLContext]
var timeout : int
var url : str
Methods
def send(self, *, text: Optional[str] = None, attachments: Optional[Sequence[Union[Dict[str, Any], Attachment]]] = None, blocks: Optional[Sequence[Union[Dict[str, Any], Block]]] = None, response_type: Optional[str] = None, replace_original: Optional[bool] = None, delete_original: Optional[bool] = None, unfurl_links: Optional[bool] = None, unfurl_media: Optional[bool] = None, metadata: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) ‑> WebhookResponse
-
Performs a Slack API request and returns the result.
Args
text
- The text message (even when having blocks, setting this as well is recommended as it works as fallback)
attachments
- A collection of attachments
blocks
- A collection of Block Kit UI components
response_type
- The type of message (either 'in_channel' or 'ephemeral')
replace_original
- True if you use this option for response_url requests
delete_original
- True if you use this option for response_url requests
unfurl_links
- Option to indicate whether text url should unfurl
unfurl_media
- Option to indicate whether media url should unfurl
metadata
- Metadata attached to the message
headers
- Request headers to append only for this request
Returns
Webhook response
Expand source code
def send( self, *, text: Optional[str] = None, attachments: Optional[Sequence[Union[Dict[str, Any], Attachment]]] = None, blocks: Optional[Sequence[Union[Dict[str, Any], Block]]] = None, response_type: Optional[str] = None, replace_original: Optional[bool] = None, delete_original: Optional[bool] = None, unfurl_links: Optional[bool] = None, unfurl_media: Optional[bool] = None, metadata: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, ) -> WebhookResponse: """Performs a Slack API request and returns the result. Args: text: The text message (even when having blocks, setting this as well is recommended as it works as fallback) attachments: A collection of attachments blocks: A collection of Block Kit UI components response_type: The type of message (either 'in_channel' or 'ephemeral') replace_original: True if you use this option for response_url requests delete_original: True if you use this option for response_url requests unfurl_links: Option to indicate whether text url should unfurl unfurl_media: Option to indicate whether media url should unfurl metadata: Metadata attached to the message headers: Request headers to append only for this request Returns: Webhook response """ return self.send_dict( # It's fine to have None value elements here # because _build_body() filters them out when constructing the actual body data body={ "text": text, "attachments": attachments, "blocks": blocks, "response_type": response_type, "replace_original": replace_original, "delete_original": delete_original, "unfurl_links": unfurl_links, "unfurl_media": unfurl_media, "metadata": metadata, }, headers=headers, )
def send_dict(self, body: Dict[str, Any], headers: Optional[Dict[str, str]] = None) ‑> WebhookResponse
-
Performs a Slack API request and returns the result.
Args
body
- JSON data structure (it's still a dict at this point), if you give this argument, body_params and files will be skipped
headers
- Request headers to append only for this request
Returns
Webhook response
Expand source code
def send_dict(self, body: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> WebhookResponse: """Performs a Slack API request and returns the result. Args: body: JSON data structure (it's still a dict at this point), if you give this argument, body_params and files will be skipped headers: Request headers to append only for this request Returns: Webhook response """ return self._perform_http_request( body=_build_body(body), headers=_build_request_headers(self.default_headers, headers), )