Module slack_sdk.web.async_base_client

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[slack_sdk.http_retry.async_handler.AsyncRetryHandler]] = 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[AsyncRetryHandler]] = None,
    ):
        self.token = None if token is None else token.strip()
        """A string specifying an `xoxp-*` or `xoxb-*` token."""
        self.base_url = base_url
        """A string representing the Slack API base URL.
        Default is `'https://www.slack.com/api/'`."""
        self.timeout = timeout
        """The maximum number of seconds the client will wait
        to connect and receive a response from Slack.
        Default is 30 seconds."""
        self.ssl = ssl
        """An [`ssl.SSLContext`](https://docs.python.org/3/library/ssl.html#ssl.SSLContext)
        instance, helpful for specifying your own custom
        certificate chain."""
        self.proxy = proxy
        """String representing a fully-qualified URL to a proxy through which
        to route all requests to the Slack API. Even if this parameter
        is not specified, if any of the following environment variables are
        present, they will be loaded into this parameter: `HTTPS_PROXY`,
        `https_proxy`, `HTTP_PROXY` or `http_proxy`."""
        self.session = session
        """An [`aiohttp.ClientSession`](https://docs.aiohttp.org/en/stable/client_reference.html#client-session)
        to attach to all outgoing requests."""
        # https://github.com/slackapi/python-slack-sdk/issues/738
        self.trust_env_in_session = trust_env_in_session
        """Boolean setting whether aiohttp outgoing requests
        are allowed to read environment variables. Commonly used in conjunction
        with proxy support via the `HTTPS_PROXY`, `https_proxy`, `HTTP_PROXY` and
        `http_proxy` environment variables."""
        self.headers = headers or {}
        """`dict` representing additional request headers to attach to all requests."""
        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: Optional[dict] = None,
        data: Union[dict, FormData] = None,
        params: Optional[dict] = None,
        json: Optional[dict] = None,  # skipcq: PYL-W0621
        headers: Optional[dict] = None,
        auth: Optional[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_deprecation_warning_if_any(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

• AsyncWebClient

Class variables

var BASE_URL

Instance variables

var base_url

A string representing the Slack API base URL. Default is 'https://www.slack.com/api/'.

var headers

dict representing additional request headers to attach to all requests.

var proxy

String representing a fully-qualified URL to a proxy through which to route all requests to the Slack API. Even if this parameter is not specified, if any of the following environment variables are present, they will be loaded into this parameter: HTTPS_PROXY, https_proxy, HTTP_PROXY or http_proxy.

var session

An aiohttp.ClientSession to attach to all outgoing requests.

var ssl

An ssl.SSLContext instance, helpful for specifying your own custom certificate chain.

var timeout

The maximum number of seconds the client will wait to connect and receive a response from Slack. Default is 30 seconds.

var token

A string specifying an xoxp-* or xoxb-* token.

var trust_env_in_session

Boolean setting whether aiohttp outgoing requests are allowed to read environment variables. Commonly used in conjunction with proxy support via the HTTPS_PROXY, https_proxy, HTTP_PROXY and http_proxy environment variables.

Methods

async def api_call(self, api_method: str, *, http_verb: str = 'POST', files: Optional[dict] = None, data: Union[dict, aiohttp.formdata.FormData] = None, params: Optional[dict] = None, json: Optional[dict] = None, headers: Optional[dict] = None, auth: Optional[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: Optional[dict] = None,
    data: Union[dict, FormData] = None,
    params: Optional[dict] = None,
    json: Optional[dict] = None,  # skipcq: PYL-W0621
    headers: Optional[dict] = None,
    auth: Optional[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_deprecation_warning_if_any(api_method)

    return await self._send(
        http_verb=http_verb,
        api_url=api_url,
        req_args=req_args,
    )