Module slack_sdk.web.async_base_client
Expand source code
import logging
from ssl import SSLContext
from typing import Optional, Union, Dict, Any, List
import aiohttp
from aiohttp import FormData, BasicAuth
from .async_internal_utils import (
_files_to_data,
_request_with_session,
)
from .async_slack_response import AsyncSlackResponse
from .deprecation import show_2020_01_deprecation
from .internal_utils import (
convert_bool_to_0_or_1,
_build_req_args,
_get_url,
get_user_agent,
)
from ..proxy_env_variable_loader import load_http_proxy_from_env
from slack_sdk.http_retry.builtin_async_handlers import async_default_handlers
from slack_sdk.http_retry.handler import RetryHandler
class AsyncBaseClient:
BASE_URL = "https://www.slack.com/api/"
def __init__(
self,
token: Optional[str] = None,
base_url: str = BASE_URL,
timeout: int = 30,
ssl: Optional[SSLContext] = None,
proxy: Optional[str] = None,
session: Optional[aiohttp.ClientSession] = None,
trust_env_in_session: bool = False,
headers: Optional[dict] = None,
user_agent_prefix: Optional[str] = None,
user_agent_suffix: Optional[str] = None,
# for Org-Wide App installation
team_id: Optional[str] = None,
logger: Optional[logging.Logger] = None,
retry_handlers: Optional[List[RetryHandler]] = None,
):
self.token = None if token is None else token.strip()
self.base_url = base_url
self.timeout = timeout
self.ssl = ssl
self.proxy = proxy
self.session = session
# https://github.com/slackapi/python-slack-sdk/issues/738
self.trust_env_in_session = trust_env_in_session
self.headers = headers or {}
self.headers["User-Agent"] = get_user_agent(
user_agent_prefix, user_agent_suffix
)
self.default_params = {}
if team_id is not None:
self.default_params["team_id"] = team_id
self._logger = logger if logger is not None else logging.getLogger(__name__)
self.retry_handlers = (
retry_handlers if retry_handlers is not None else async_default_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
async def api_call( # skipcq: PYL-R1710
self,
api_method: str,
*,
http_verb: str = "POST",
files: dict = None,
data: Union[dict, FormData] = None,
params: dict = None,
json: dict = None, # skipcq: PYL-W0621
headers: dict = None,
auth: dict = None,
) -> AsyncSlackResponse:
"""Create a request and execute the API call to Slack.
Args:
api_method (str): The target Slack API method.
e.g. 'chat.postMessage'
http_verb (str): HTTP Verb. e.g. 'POST'
files (dict): Files to multipart upload.
e.g. {image OR file: file_object OR file_path}
data: The body to attach to the request. If a dictionary is
provided, form-encoding will take place.
e.g. {'key1': 'value1', 'key2': 'value2'}
params (dict): The URL parameters to append to the URL.
e.g. {'key1': 'value1', 'key2': 'value2'}
json (dict): JSON for the body to attach to the request
(if files or data is not specified).
e.g. {'key1': 'value1', 'key2': 'value2'}
headers (dict): Additional request headers
auth (dict): A dictionary that consists of client_id and client_secret
Returns:
(AsyncSlackResponse)
The server's response to an HTTP request. Data
from the response can be accessed like a dict.
If the response included 'next_cursor' it can
be iterated on to execute subsequent requests.
Raises:
SlackApiError: The following Slack API call failed:
'chat.postMessage'.
SlackRequestError: Json data can only be submitted as
POST requests.
"""
api_url = _get_url(self.base_url, api_method)
if auth is not None:
if isinstance(auth, dict):
auth = BasicAuth(auth["client_id"], auth["client_secret"])
if isinstance(auth, BasicAuth):
if headers is None:
headers = {}
headers["Authorization"] = auth.encode()
auth = None
headers = headers or {}
headers.update(self.headers)
req_args = _build_req_args(
token=self.token,
http_verb=http_verb,
files=files,
data=data,
default_params=self.default_params,
params=params,
json=json, # skipcq: PYL-W0621
headers=headers,
auth=auth,
ssl=self.ssl,
proxy=self.proxy,
)
show_2020_01_deprecation(api_method)
return await self._send(
http_verb=http_verb,
api_url=api_url,
req_args=req_args,
)
async def _send(
self, http_verb: str, api_url: str, req_args: dict
) -> AsyncSlackResponse:
"""Sends the request out for transmission.
Args:
http_verb (str): The HTTP verb. e.g. 'GET' or 'POST'.
api_url (str): The Slack API url. e.g. 'https://slack.com/api/chat.postMessage'
req_args (dict): The request arguments to be attached to the request.
e.g.
{
json: {
'attachments': [{"pretext": "pre-hello", "text": "text-world"}],
'channel': '#random'
}
}
Returns:
The response parsed into a AsyncSlackResponse object.
"""
open_files = _files_to_data(req_args)
try:
if "params" in req_args:
# True/False -> "1"/"0"
req_args["params"] = convert_bool_to_0_or_1(req_args["params"])
res = await self._request(
http_verb=http_verb, api_url=api_url, req_args=req_args
)
finally:
for f in open_files:
f.close()
data = {
"client": self,
"http_verb": http_verb,
"api_url": api_url,
"req_args": req_args,
}
return AsyncSlackResponse(**{**data, **res}).validate()
async def _request(self, *, http_verb, api_url, req_args) -> Dict[str, Any]:
"""Submit the HTTP request with the running session or a new session.
Returns:
A dictionary of the response data.
"""
return await _request_with_session(
current_session=self.session,
timeout=self.timeout,
logger=self._logger,
http_verb=http_verb,
api_url=api_url,
req_args=req_args,
retry_handlers=self.retry_handlers,
)
Classes
class AsyncBaseClient (token: Optional[str] = None, base_url: str = 'https://www.slack.com/api/', timeout: int = 30, ssl: Optional[ssl.SSLContext] = None, proxy: Optional[str] = None, session: Optional[aiohttp.client.ClientSession] = None, trust_env_in_session: bool = False, headers: Optional[dict] = None, user_agent_prefix: Optional[str] = None, user_agent_suffix: Optional[str] = None, team_id: Optional[str] = None, logger: Optional[logging.Logger] = None, retry_handlers: Optional[List[RetryHandler]] = None)
-
Expand source code
class AsyncBaseClient: BASE_URL = "https://www.slack.com/api/" def __init__( self, token: Optional[str] = None, base_url: str = BASE_URL, timeout: int = 30, ssl: Optional[SSLContext] = None, proxy: Optional[str] = None, session: Optional[aiohttp.ClientSession] = None, trust_env_in_session: bool = False, headers: Optional[dict] = None, user_agent_prefix: Optional[str] = None, user_agent_suffix: Optional[str] = None, # for Org-Wide App installation team_id: Optional[str] = None, logger: Optional[logging.Logger] = None, retry_handlers: Optional[List[RetryHandler]] = None, ): self.token = None if token is None else token.strip() self.base_url = base_url self.timeout = timeout self.ssl = ssl self.proxy = proxy self.session = session # https://github.com/slackapi/python-slack-sdk/issues/738 self.trust_env_in_session = trust_env_in_session self.headers = headers or {} self.headers["User-Agent"] = get_user_agent( user_agent_prefix, user_agent_suffix ) self.default_params = {} if team_id is not None: self.default_params["team_id"] = team_id self._logger = logger if logger is not None else logging.getLogger(__name__) self.retry_handlers = ( retry_handlers if retry_handlers is not None else async_default_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 async def api_call( # skipcq: PYL-R1710 self, api_method: str, *, http_verb: str = "POST", files: dict = None, data: Union[dict, FormData] = None, params: dict = None, json: dict = None, # skipcq: PYL-W0621 headers: dict = None, auth: dict = None, ) -> AsyncSlackResponse: """Create a request and execute the API call to Slack. Args: api_method (str): The target Slack API method. e.g. 'chat.postMessage' http_verb (str): HTTP Verb. e.g. 'POST' files (dict): Files to multipart upload. e.g. {image OR file: file_object OR file_path} data: The body to attach to the request. If a dictionary is provided, form-encoding will take place. e.g. {'key1': 'value1', 'key2': 'value2'} params (dict): The URL parameters to append to the URL. e.g. {'key1': 'value1', 'key2': 'value2'} json (dict): JSON for the body to attach to the request (if files or data is not specified). e.g. {'key1': 'value1', 'key2': 'value2'} headers (dict): Additional request headers auth (dict): A dictionary that consists of client_id and client_secret Returns: (AsyncSlackResponse) The server's response to an HTTP request. Data from the response can be accessed like a dict. If the response included 'next_cursor' it can be iterated on to execute subsequent requests. Raises: SlackApiError: The following Slack API call failed: 'chat.postMessage'. SlackRequestError: Json data can only be submitted as POST requests. """ api_url = _get_url(self.base_url, api_method) if auth is not None: if isinstance(auth, dict): auth = BasicAuth(auth["client_id"], auth["client_secret"]) if isinstance(auth, BasicAuth): if headers is None: headers = {} headers["Authorization"] = auth.encode() auth = None headers = headers or {} headers.update(self.headers) req_args = _build_req_args( token=self.token, http_verb=http_verb, files=files, data=data, default_params=self.default_params, params=params, json=json, # skipcq: PYL-W0621 headers=headers, auth=auth, ssl=self.ssl, proxy=self.proxy, ) show_2020_01_deprecation(api_method) return await self._send( http_verb=http_verb, api_url=api_url, req_args=req_args, ) async def _send( self, http_verb: str, api_url: str, req_args: dict ) -> AsyncSlackResponse: """Sends the request out for transmission. Args: http_verb (str): The HTTP verb. e.g. 'GET' or 'POST'. api_url (str): The Slack API url. e.g. 'https://slack.com/api/chat.postMessage' req_args (dict): The request arguments to be attached to the request. e.g. { json: { 'attachments': [{"pretext": "pre-hello", "text": "text-world"}], 'channel': '#random' } } Returns: The response parsed into a AsyncSlackResponse object. """ open_files = _files_to_data(req_args) try: if "params" in req_args: # True/False -> "1"/"0" req_args["params"] = convert_bool_to_0_or_1(req_args["params"]) res = await self._request( http_verb=http_verb, api_url=api_url, req_args=req_args ) finally: for f in open_files: f.close() data = { "client": self, "http_verb": http_verb, "api_url": api_url, "req_args": req_args, } return AsyncSlackResponse(**{**data, **res}).validate() async def _request(self, *, http_verb, api_url, req_args) -> Dict[str, Any]: """Submit the HTTP request with the running session or a new session. Returns: A dictionary of the response data. """ return await _request_with_session( current_session=self.session, timeout=self.timeout, logger=self._logger, http_verb=http_verb, api_url=api_url, req_args=req_args, retry_handlers=self.retry_handlers, )
Subclasses
Class variables
var BASE_URL
Methods
async def api_call(self, api_method: str, *, http_verb: str = 'POST', files: dict = None, data: Union[dict, aiohttp.formdata.FormData] = None, params: dict = None, json: dict = None, headers: dict = None, auth: dict = None) ‑> AsyncSlackResponse
-
Create a request and execute the API call to Slack.
Args
api_method
:str
- The target Slack API method. e.g. 'chat.postMessage'
http_verb
:str
- HTTP Verb. e.g. 'POST'
files
:dict
- Files to multipart upload. e.g. {image OR file: file_object OR file_path}
data
- The body to attach to the request. If a dictionary is provided, form-encoding will take place. e.g. {'key1': 'value1', 'key2': 'value2'}
params
:dict
- The URL parameters to append to the URL. e.g. {'key1': 'value1', 'key2': 'value2'}
json
:dict
- JSON for the body to attach to the request (if files or data is not specified). e.g. {'key1': 'value1', 'key2': 'value2'}
headers
:dict
- Additional request headers
auth
:dict
- A dictionary that consists of client_id and client_secret
Returns
(AsyncSlackResponse) The server's response to an HTTP request. Data from the response can be accessed like a dict. If the response included 'next_cursor' it can be iterated on to execute subsequent requests.
Raises
SlackApiError
- The following Slack API call failed: 'chat.postMessage'.
SlackRequestError
- Json data can only be submitted as POST requests.
Expand source code
async def api_call( # skipcq: PYL-R1710 self, api_method: str, *, http_verb: str = "POST", files: dict = None, data: Union[dict, FormData] = None, params: dict = None, json: dict = None, # skipcq: PYL-W0621 headers: dict = None, auth: dict = None, ) -> AsyncSlackResponse: """Create a request and execute the API call to Slack. Args: api_method (str): The target Slack API method. e.g. 'chat.postMessage' http_verb (str): HTTP Verb. e.g. 'POST' files (dict): Files to multipart upload. e.g. {image OR file: file_object OR file_path} data: The body to attach to the request. If a dictionary is provided, form-encoding will take place. e.g. {'key1': 'value1', 'key2': 'value2'} params (dict): The URL parameters to append to the URL. e.g. {'key1': 'value1', 'key2': 'value2'} json (dict): JSON for the body to attach to the request (if files or data is not specified). e.g. {'key1': 'value1', 'key2': 'value2'} headers (dict): Additional request headers auth (dict): A dictionary that consists of client_id and client_secret Returns: (AsyncSlackResponse) The server's response to an HTTP request. Data from the response can be accessed like a dict. If the response included 'next_cursor' it can be iterated on to execute subsequent requests. Raises: SlackApiError: The following Slack API call failed: 'chat.postMessage'. SlackRequestError: Json data can only be submitted as POST requests. """ api_url = _get_url(self.base_url, api_method) if auth is not None: if isinstance(auth, dict): auth = BasicAuth(auth["client_id"], auth["client_secret"]) if isinstance(auth, BasicAuth): if headers is None: headers = {} headers["Authorization"] = auth.encode() auth = None headers = headers or {} headers.update(self.headers) req_args = _build_req_args( token=self.token, http_verb=http_verb, files=files, data=data, default_params=self.default_params, params=params, json=json, # skipcq: PYL-W0621 headers=headers, auth=auth, ssl=self.ssl, proxy=self.proxy, ) show_2020_01_deprecation(api_method) return await self._send( http_verb=http_verb, api_url=api_url, req_args=req_args, )