Module `slack_bolt.middleware.async_builtins`

Expand source code

from .ignoring_self_events.async_ignoring_self_events import (
    AsyncIgnoringSelfEvents,
)
from .request_verification.async_request_verification import (
    AsyncRequestVerification,
)
from .ssl_check.async_ssl_check import AsyncSslCheck
from .url_verification.async_url_verification import AsyncUrlVerification
from .message_listener_matches.async_message_listener_matches import (
    AsyncMessageListenerMatches,
)

__all__ = [
    "AsyncIgnoringSelfEvents",
    "AsyncRequestVerification",
    "AsyncSslCheck",
    "AsyncUrlVerification",
    "AsyncMessageListenerMatches",
]

Classes

class AsyncIgnoringSelfEvents (base_logger: Optional[logging.Logger] = None)

A middleware can process request data before other middleware and listener functions.

Ignores the events generated by this bot user itself.

Expand source code

class AsyncIgnoringSelfEvents(IgnoringSelfEvents, AsyncMiddleware):
    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        auth_result = req.context.authorize_result
        # message events can have $.event.bot_id while it does not have its user_id
        bot_id = req.body.get("event", {}).get("bot_id")
        if self._is_self_event(auth_result, req.context.user_id, bot_id, req.body):
            self._debug_log(req.body)
            return await req.context.ack()
        else:
            return await next()

Ancestors

Inherited members

IgnoringSelfEvents:
- name
- process
AsyncMiddleware:
- async_process

class AsyncMessageListenerMatches (keyword: Union[str, Pattern])

A middleware can process request data before other middleware and listener functions.

Captures matched keywords and saves the values in context.

Expand source code

class AsyncMessageListenerMatches(AsyncMiddleware):
    def __init__(self, keyword: Union[str, Pattern]):
        """Captures matched keywords and saves the values in context."""
        self.keyword = keyword

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        text = req.body.get("event", {}).get("text", "")
        if text:
            m = re.findall(self.keyword, text)
            if m is not None and m != []:
                if type(m[0]) is not tuple:
                    m = tuple(m)
                else:
                    m = m[0]
                req.context["matches"] = m  # tuple or list
                return await next()

        # As the text doesn't match, skip running the listener
        return resp

Ancestors

AsyncMiddleware

Inherited members

AsyncMiddleware:
- async_process
- name

class AsyncRequestVerification (signing_secret: str, base_logger: Optional[logging.Logger] = None)

Verifies an incoming request by checking the validity of x-slack-signature, x-slack-request-timestamp, and its body data.

Refer to https://api.slack.com/authentication/verifying-requests-from-slack for details.

Verifies an incoming request by checking the validity of x-slack-signature, x-slack-request-timestamp, and its body data.

Refer to https://api.slack.com/authentication/verifying-requests-from-slack for details.

Args

signing_secret: The signing secret
base_logger: The base logger

Expand source code

class AsyncRequestVerification(RequestVerification, AsyncMiddleware):
    """Verifies an incoming request by checking the validity of
    `x-slack-signature`, `x-slack-request-timestamp`, and its body data.

    Refer to https://api.slack.com/authentication/verifying-requests-from-slack for details.
    """

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._can_skip(req.mode, req.body):
            return await next()

        body = req.raw_body
        timestamp = req.headers.get("x-slack-request-timestamp", ["0"])[0]
        signature = req.headers.get("x-slack-signature", [""])[0]
        if self.verifier.is_valid(body, timestamp, signature):
            return await next()
        else:
            self._debug_log_error(signature, timestamp, body)
            return self._build_error_response()

Ancestors

Inherited members

RequestVerification:
- name
- process
AsyncMiddleware:
- async_process

class AsyncSslCheck (verification_token: Optional[str] = None, base_logger: Optional[logging.Logger] = None)

A middleware can process request data before other middleware and listener functions.

Handles ssl_check requests. Refer to https://api.slack.com/interactivity/slash-commands for details.

Args

verification_token: The verification token to check (optional as it's already deprecated - https://api.slack.com/authentication/verifying-requests-from-slack#verification_token_deprecation)
base_logger: The base logger

Expand source code

class AsyncSslCheck(SslCheck, AsyncMiddleware):
    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._is_ssl_check_request(req.body):
            if self._verify_token_if_needed(req.body):
                return self._build_error_response()
            return self._build_success_response()
        else:
            return await next()

Ancestors

Class variables

var logger : logging.Logger
var verification_token : Optional[str]

Inherited members

SslCheck:
- name
- process
AsyncMiddleware:
- async_process

class AsyncUrlVerification (base_logger: Optional[logging.Logger] = None)

A middleware can process request data before other middleware and listener functions.

Handles url_verification requests.

Refer to https://api.slack.com/events/url_verification for details.

Args

base_logger: The base logger

Expand source code

class AsyncUrlVerification(UrlVerification, AsyncMiddleware):
    def __init__(self, base_logger: Optional[Logger] = None):
        self.logger = get_bolt_logger(AsyncUrlVerification, base_logger=base_logger)

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._is_url_verification_request(req.body):
            return self._build_success_response(req.body)
        else:
            return await next()

Ancestors

Inherited members

UrlVerification:
- name
- process
AsyncMiddleware:
- async_process